This is an automated email from the ASF dual-hosted git repository.
lgbo-ustc pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 9685185b94 [GLUTEN-12206][FLINK] Support native checkpoint snapshot
and restore for source operator (#12490)
9685185b94 is described below
commit 9685185b9478b4ba1071fd2f03204942284bcad5
Author: kevinyhzou <[email protected]>
AuthorDate: Mon Jul 13 09:40:30 2026 +0800
[GLUTEN-12206][FLINK] Support native checkpoint snapshot and restore for
source operator (#12490)
* feat: persist native source checkpoint state
* fix: sync Gluten task output metrics
fix: keep Gluten compatible with older velox4j APIs
Revert "fix: keep Gluten compatible with older velox4j APIs"
This reverts commit 67a3cccbf6013af0e91c3a5b56e85bc4d869af2b.
[FLINK] Drain Velox pipeline in prepareSnapshotPreBarrier for checkpoint
consistency
remove changes in source metrics
remove metric changes
remove metrics
* update fink.md
---------
Co-authored-by: zhangzhibiao <[email protected]>
---
.github/workflows/flink.yml | 4 +--
gluten-flink/docs/Flink.md | 2 +-
...TaskMetrics.java => SourceOperatorMetrics.java} | 4 +--
.../runtime/operators/GlutenOneInputOperator.java | 3 +-
.../runtime/operators/GlutenSourceFunction.java | 35 +++++++++++++++++-----
.../org/apache/gluten/util/VectorOutputBridge.java | 10 ++++---
6 files changed, 39 insertions(+), 19 deletions(-)
diff --git a/.github/workflows/flink.yml b/.github/workflows/flink.yml
index b66bddf427..e32cc1b7fc 100644
--- a/.github/workflows/flink.yml
+++ b/.github/workflows/flink.yml
@@ -87,8 +87,8 @@ jobs:
export VELOX_DEPENDENCY_SOURCE=BUNDLED
export fmt_SOURCE=BUNDLED
export folly_SOURCE=BUNDLED
- git clone -b feature/watermark-status-propagation
https://github.com/bigo-sg/velox4j.git
- cd velox4j && git reset --hard 6e2046f
+ git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
+ cd velox4j && git reset --hard
3ccc4e61dec4d569dc3dd9408b78ee1d4facf3ba
git apply $GITHUB_WORKSPACE/gluten-flink/patches/fix-velox4j.patch
$GITHUB_WORKSPACE/build/mvn clean install -DskipTests -Dgpg.skip
-Dspotless.skip=true
cd ..
diff --git a/gluten-flink/docs/Flink.md b/gluten-flink/docs/Flink.md
index 9c1900f06b..3e3adde696 100644
--- a/gluten-flink/docs/Flink.md
+++ b/gluten-flink/docs/Flink.md
@@ -48,7 +48,7 @@ As some features have not been committed to upstream, you
have to use the follow
## fetch velox4j code
git clone -b gluten-0530 https://github.com/bigo-sg/velox4j.git
cd velox4j
-git reset --hard 1b1ad0833220476be6a042b83a474cf58f8f076c
+git reset --hard 3ccc4e61dec4d569dc3dd9408b78ee1d4facf3ba
mvn clean install -DskipTests -Dgpg.skip -Dspotless.skip=true
```
**Get gluten**
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/metrics/SourceTaskMetrics.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/metrics/SourceOperatorMetrics.java
similarity index 96%
rename from
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/metrics/SourceTaskMetrics.java
rename to
gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/metrics/SourceOperatorMetrics.java
index 5bbbb028b6..8dcc80438f 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/metrics/SourceTaskMetrics.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/metrics/SourceOperatorMetrics.java
@@ -24,7 +24,7 @@ import org.apache.flink.metrics.groups.OperatorMetricGroup;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
-public class SourceTaskMetrics {
+public class SourceOperatorMetrics {
private final String keyOperatorType = "operatorType";
private final String sourceOperatorName = "TableScan";
@@ -35,7 +35,7 @@ public class SourceTaskMetrics {
private Counter sourceNumBytesOut;
private long lastUpdateTime = System.currentTimeMillis();
- public SourceTaskMetrics(OperatorMetricGroup metricGroup) {
+ public SourceOperatorMetrics(OperatorMetricGroup metricGroup) {
sourceNumRecordsOut =
metricGroup.getIOMetricGroup().getNumRecordsOutCounter();
sourceNumBytesOut = metricGroup.getIOMetricGroup().getNumBytesOutCounter();
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
index 5729a60b41..ba9329e67b 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenOneInputOperator.java
@@ -326,14 +326,13 @@ public class GlutenOneInputOperator<IN, OUT> extends
TableStreamOperator<OUT>
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
- // TODO: notify velox
super.prepareSnapshotPreBarrier(checkpointId);
}
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
// TODO: implement it
- task.snapshotState(0);
+ task.snapshotState(context.getCheckpointId());
super.snapshotState(context);
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
index 5dfc8447a9..a47b6c7c17 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/table/runtime/operators/GlutenSourceFunction.java
@@ -18,7 +18,7 @@ package org.apache.gluten.table.runtime.operators;
import org.apache.gluten.table.runtime.config.VeloxConnectorConfig;
import org.apache.gluten.table.runtime.config.VeloxQueryConfig;
-import org.apache.gluten.table.runtime.metrics.SourceTaskMetrics;
+import org.apache.gluten.table.runtime.metrics.SourceOperatorMetrics;
import org.apache.gluten.vectorized.FlinkRowToVLVectorConvertor;
import io.github.zhztheplayer.velox4j.connector.ConnectorSplit;
@@ -33,6 +33,8 @@ import io.github.zhztheplayer.velox4j.stateful.StatefulRecord;
import io.github.zhztheplayer.velox4j.stateful.StatefulWatermark;
import io.github.zhztheplayer.velox4j.type.RowType;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
@@ -44,6 +46,7 @@ import org.apache.flink.table.data.RowData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -64,8 +67,10 @@ public class GlutenSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
private GlutenSessionResource sessionResource;
private Query query;
private SerialTask task;
- private SourceTaskMetrics taskMetrics;
+ private SourceOperatorMetrics metrics;
private final Class<OUT> outClass;
+ private transient ListState<String> checkpointState;
+ private transient String[] restoredCheckpointRecords = new String[0];
public GlutenSourceFunction(
StatefulPlanNode planNode,
@@ -115,7 +120,7 @@ public class GlutenSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
default:
return;
}
- taskMetrics.updateMetrics(task, id);
+ metrics.updateMetrics(task, id);
}
}
@@ -208,15 +213,29 @@ public class GlutenSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
- // TODO: implement it
- this.task.snapshotState(0);
+ checkpointState.clear();
+ String[] checkpointRecords =
this.task.snapshotState(context.getCheckpointId());
+ for (String checkpointRecord : checkpointRecords) {
+ checkpointState.add(checkpointRecord);
+ }
}
@Override
public void initializeState(FunctionInitializationContext context) throws
Exception {
+ checkpointState =
+ context
+ .getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>("gluten-native-source-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]);
+ }
initSession();
- // TODO: implement it
- this.task.initializeState(0, null);
+ this.task.initializeState(0, null, restoredCheckpointRecords);
}
public String[] notifyCheckpointComplete(long checkpointId) throws Exception
{
@@ -252,6 +271,6 @@ public class GlutenSourceFunction<OUT> extends
RichParallelSourceFunction<OUT>
task = session.queryOps().execute(query);
task.addSplit(id, activeSplit);
task.noMoreSplits(id);
- taskMetrics = new SourceTaskMetrics(getRuntimeContext().getMetricGroup());
+ metrics = new SourceOperatorMetrics(getRuntimeContext().getMetricGroup());
}
}
diff --git
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
index bbc4ea796c..0e749daa82 100644
---
a/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
+++
b/gluten-flink/runtime/src/main/java/org/apache/gluten/util/VectorOutputBridge.java
@@ -41,7 +41,7 @@ public interface VectorOutputBridge<OUT> extends Serializable
{
* @param allocator buffer allocator for converting RowVector
* @param outputType the RowType schema of the output
*/
- void collect(
+ long collect(
Output<StreamRecord<OUT>> collector,
StatefulRecord record,
BufferAllocator allocator,
@@ -91,7 +91,7 @@ public interface VectorOutputBridge<OUT> extends Serializable
{
}
@Override
- public void collect(
+ public long collect(
Output<StreamRecord<org.apache.flink.table.data.RowData>> collector,
StatefulRecord record,
BufferAllocator allocator,
@@ -102,6 +102,7 @@ public interface VectorOutputBridge<OUT> extends
Serializable {
for (org.apache.flink.table.data.RowData row : rows) {
collector.collect(getOrCreateOutputElement().replace(row));
}
+ return rows.size();
}
private StreamRecord<org.apache.flink.table.data.RowData>
getOrCreateOutputElement() {
@@ -125,12 +126,13 @@ public interface VectorOutputBridge<OUT> extends
Serializable {
}
@Override
- public void collect(
+ public long collect(
Output<StreamRecord<StatefulRecord>> collector,
StatefulRecord record,
BufferAllocator allocator,
RowType outputType) {
collector.collect(getOrCreateOutputElement().replace(record));
+ return 1;
}
private StreamRecord<StatefulRecord> getOrCreateOutputElement() {
@@ -144,7 +146,7 @@ public interface VectorOutputBridge<OUT> extends
Serializable {
class PartitionCommitInfoOutputBridge implements
VectorOutputBridge<PartitionCommitInfo> {
@Override
- public void collect(
+ public long collect(
Output<StreamRecord<PartitionCommitInfo>> collector,
StatefulRecord record,
BufferAllocator allocator,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
