From dbbd3544f9cac4a8204eeb2897db0b36e7195429 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Wed, 17 Jun 2026 20:54:17 +0800 Subject: [PATCH 01/10] feat: propagate source checkpoint records --- src/main/cpp/CMakeLists.txt | 2 +- src/main/cpp/main/velox4j/jni/JniWrapper.cc | 37 ++++++++++++++++--- .../velox4j/query/StatefulQueryExecutor.cc | 11 +++--- .../velox4j/query/StatefulQueryExecutor.h | 6 ++- src/main/cpp/velox-ref.txt | 2 +- .../zhztheplayer/velox4j/jni/JniApi.java | 12 ++++-- .../zhztheplayer/velox4j/jni/JniWrapper.java | 5 ++- .../velox4j/query/SerialTask.java | 13 +++++-- 8 files changed, 64 insertions(+), 24 deletions(-) diff --git a/src/main/cpp/CMakeLists.txt b/src/main/cpp/CMakeLists.txt index 656fe1ef11..b12ee48f7a 100644 --- a/src/main/cpp/CMakeLists.txt +++ b/src/main/cpp/CMakeLists.txt @@ -74,7 +74,7 @@ cpmaddpackage( GIT_REPOSITORY ${VELOX_SOURCE_URL} GIT_TAG - be3e18f53a2cc655bb743c8bf2afb312dbd40ee4) + ${VELOX_REF}) # Import JniHelpers. set(JNI_HELPERS_REPO zhztheplayer/JniHelpersCpp) diff --git a/src/main/cpp/main/velox4j/jni/JniWrapper.cc b/src/main/cpp/main/velox4j/jni/JniWrapper.cc index 4fe69ba097..2bde630341 100644 --- a/src/main/cpp/main/velox4j/jni/JniWrapper.cc +++ b/src/main/cpp/main/velox4j/jni/JniWrapper.cc @@ -201,19 +201,43 @@ void initializeState( jobject javaThis, jlong itrId, jlong context, - jstring keyedStateBackendConfigString) { + jstring keyedStateBackendConfigString, + jobjectArray checkpointRecords) { JNI_METHOD_START auto itr = ObjectStore::retrieve(itrId); spotify::jni::JavaString jTypeJson{env, keyedStateBackendConfigString}; - itr->initializeState(context, jTypeJson.get()); + std::vector records; + if (checkpointRecords != nullptr) { + const auto numRecords = env->GetArrayLength(checkpointRecords); + records.reserve(numRecords); + for (jsize i = 0; i < numRecords; ++i) { + auto record = static_cast( + env->GetObjectArrayElement(checkpointRecords, i)); + spotify::jni::JavaString jRecord{env, record}; + records.push_back(jRecord.get()); + env->DeleteLocalRef(record); + } + } + itr->initializeState(context, jTypeJson.get(), std::move(records)); JNI_METHOD_END() } -void snapshotState(JNIEnv* env, jobject javaThis, jlong itrId, jlong context) { +jobject +snapshotState(JNIEnv* env, jobject javaThis, jlong itrId, jlong context) { JNI_METHOD_START auto itr = ObjectStore::retrieve(itrId); - itr->snapshotState(context); - JNI_METHOD_END() + std::vector snapshots = itr->snapshotState(context); + jclass stringClass = env->FindClass("java/lang/String"); + jobjectArray result = + env->NewObjectArray(snapshots.size(), stringClass, nullptr); + for (size_t i = 0; i < snapshots.size(); ++i) { + jstring str = env->NewStringUTF(snapshots[i].c_str()); + env->SetObjectArrayElement(result, i, str); + env->DeleteLocalRef(str); + } + env->DeleteLocalRef(stringClass); + return result; + JNI_METHOD_END(nullptr) } jobject notifyCheckpointComplete( @@ -527,11 +551,12 @@ void JniWrapper::initialize(JNIEnv* env) { kTypeLong, kTypeLong, kTypeString, + "[Ljava/lang/String;", nullptr); addNativeMethod( "snapshotState", (void*)snapshotState, - kTypeVoid, + "[Ljava/lang/String;", kTypeLong, kTypeLong, nullptr); diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc index ecc8c8df63..686e44c59d 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc @@ -108,7 +108,8 @@ void StatefulSerialTask::notifyWatermark(long watermark) { void StatefulSerialTask::initializeState( long checkpointId, - std::string keyedStateBackendConfigString) { + std::string keyedStateBackendConfigString, + std::vector checkpointRecords) { folly::dynamic obj = folly::parseJson(keyedStateBackendConfigString); std::shared_ptr params = stateful::KeyedStateBackendParameters::create(obj, nullptr); @@ -116,15 +117,15 @@ void StatefulSerialTask::initializeState( params->getBackendType() == stateful::StateBackendType::ROCKSDB) { auto rocksdbParams = stateful::RocksDBKeyedStateBackendParameters::create(obj, nullptr); - task_->initializeState(rocksdbParams); + task_->initializeState(rocksdbParams, checkpointRecords); } else { // params maybe null, then initialize by using default heap state backend. - task_->initializeState(params); + task_->initializeState(params, checkpointRecords); } } -void StatefulSerialTask::snapshotState(long checkpointId) { - task_->snapshotState(); +std::vector StatefulSerialTask::snapshotState(long checkpointId) { + return task_->snapshotState(checkpointId); } std::vector StatefulSerialTask::notifyCheckpointComplete( diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h index d2f17e3e95..682ccd0db0 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h @@ -20,6 +20,7 @@ #include #include #include +#include #include "Query.h" #include "velox4j/iterator/UpIterator.h" #include "velox4j/memory/MemoryManager.h" @@ -49,9 +50,10 @@ class StatefulSerialTask : public UpIterator { void initializeState( long checkpointId, - std::string keyedStateBackendConfigString); + std::string keyedStateBackendConfigString, + std::vector checkpointRecords); - void snapshotState(long checkpointId); + std::vector snapshotState(long checkpointId); std::vector notifyCheckpointComplete(long checkpointId); diff --git a/src/main/cpp/velox-ref.txt b/src/main/cpp/velox-ref.txt index 93796606af..214e63a999 100644 --- a/src/main/cpp/velox-ref.txt +++ b/src/main/cpp/velox-ref.txt @@ -1 +1 @@ -92316a335ee786372b9274f87a456192bb5ee474 +72caa1c04d78db22f53274dfb6d3fc3211764833 diff --git a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java index a65ca2484e..49f7c65c49 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java @@ -105,12 +105,16 @@ public void notifyWatermark(UpIterator itr, long watermark) { jni.notifyWatermark(itr.id(), watermark); } - public void initializeState(UpIterator itr, long context, String keyedStateBackendConfigString) { - jni.initializeState(itr.id(), context, keyedStateBackendConfigString); + public void initializeState( + UpIterator itr, + long context, + String keyedStateBackendConfigString, + String[] checkpointRecords) { + jni.initializeState(itr.id(), context, keyedStateBackendConfigString, checkpointRecords); } - public void snapshotState(UpIterator itr, long context) { - jni.snapshotState(itr.id(), context); + public String[] snapshotState(UpIterator itr, long context) { + return jni.snapshotState(itr.id(), context); } public String[] notifyCheckpointComplete(UpIterator itr, long checkpointId) { diff --git a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java index fc2560a0e2..a5c8fd2137 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java @@ -58,9 +58,10 @@ public long sessionId() { // For Flink. native void notifyWatermark(long id, long watermark); - native void initializeState(long id, long context, String keyedStateBackendConfigString); + native void initializeState( + long id, long context, String keyedStateBackendConfigString, String[] checkpointRecords); - native void snapshotState(long id, long context); + native String[] snapshotState(long id, long context); native String[] notifyCheckpointComplete(long id, long checkpointId); diff --git a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java index f24d313cd4..206db34a18 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java @@ -83,13 +83,20 @@ public SerialTaskStats collectStats() { public void initializeState( long context, KeyedStateBackendParameters keyedStateBackendParameters) { + initializeState(context, keyedStateBackendParameters, new String[0]); + } + + public void initializeState( + long context, + KeyedStateBackendParameters keyedStateBackendParameters, + String[] checkpointRecords) { String keyedStateBackendJsonParameters = keyedStateBackendParameters != null ? Serde.toJson(keyedStateBackendParameters) : "{}"; - jniApi.initializeState(this, context, keyedStateBackendJsonParameters); + jniApi.initializeState(this, context, keyedStateBackendJsonParameters, checkpointRecords); } - public void snapshotState(long context) { - jniApi.snapshotState(this, context); + public String[] snapshotState(long context) { + return jniApi.snapshotState(this, context); } public String[] notifyCheckpointComplete(long checkpointId) { From 2bc5a2cb2588631a2e07097936383fff1fd271bf Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Thu, 18 Jun 2026 14:21:12 +0800 Subject: [PATCH 02/10] feat: expose all plan stats --- .../velox4j/query/SerialTaskStats.java | 8 ++++ .../velox4j/query/SerialTaskStatsTest.java | 39 +++++++++++++++++++ 2 files changed, 47 insertions(+) create mode 100644 src/test/java/io/github/zhztheplayer/velox4j/query/SerialTaskStatsTest.java diff --git a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTaskStats.java b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTaskStats.java index 833cea43a0..f3758a8fa3 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTaskStats.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTaskStats.java @@ -62,4 +62,12 @@ public ObjectNode planStats(String planNodeId) { } return out.get(0); } + + public List planStats() { + final List out = new ArrayList<>(); + for (JsonNode each : planStatsDynamic) { + out.add((ObjectNode) each); + } + return out; + } } diff --git a/src/test/java/io/github/zhztheplayer/velox4j/query/SerialTaskStatsTest.java b/src/test/java/io/github/zhztheplayer/velox4j/query/SerialTaskStatsTest.java new file mode 100644 index 0000000000..5b08531e19 --- /dev/null +++ b/src/test/java/io/github/zhztheplayer/velox4j/query/SerialTaskStatsTest.java @@ -0,0 +1,39 @@ +/* +* Licensed to the Apache Software Foundation (ASF) under one or more +* contributor license agreements. See the NOTICE file distributed with +* this work for additional information regarding copyright ownership. +* The ASF licenses this file to You under the Apache License, Version 2.0 +* (the "License"); you may not use this file except in compliance with +* the License. You may obtain a copy of the License at +* +* http://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, software +* distributed under the License is distributed on an "AS IS" BASIS, +* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +* See the License for the specific language governing permissions and +* limitations under the License. +*/ +package io.github.zhztheplayer.velox4j.query; + +import org.junit.Assert; +import org.junit.Test; + +public class SerialTaskStatsTest { + + @Test + public void testPlanStatsReturnsAllPlanStats() { + final SerialTaskStats stats = + SerialTaskStats.fromJson( + "{" + + "\"planStats\":[" + + "{\"planNodeId\":\"scan-1\",\"operatorType\":\"TableScan\"}," + + "{\"planNodeId\":\"project-1\",\"operatorType\":\"FilterProject\"}" + + "]" + + "}"); + + Assert.assertEquals(2, stats.planStats().size()); + Assert.assertEquals("scan-1", stats.planStats().get(0).get("planNodeId").asText()); + Assert.assertEquals("project-1", stats.planStats().get(1).get("planNodeId").asText()); + } +} From 5bfa9e692ddee42489781d17121ead506751f622 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Mon, 22 Jun 2026 12:30:57 +0800 Subject: [PATCH 03/10] Update Velox reference --- src/main/cpp/velox-ref.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/cpp/velox-ref.txt b/src/main/cpp/velox-ref.txt index 214e63a999..8be72431b7 100644 --- a/src/main/cpp/velox-ref.txt +++ b/src/main/cpp/velox-ref.txt @@ -1 +1 @@ -72caa1c04d78db22f53274dfb6d3fc3211764833 +01834e806060609c0557e7d66b5a5b63f0f4b469 From 8ceb4e5e108588d92b89498bb3f1a3c0c4b029b5 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Tue, 23 Jun 2026 14:25:51 +0800 Subject: [PATCH 04/10] Fix Velox checkpoint ref hash --- src/main/cpp/velox-ref.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/cpp/velox-ref.txt b/src/main/cpp/velox-ref.txt index 8be72431b7..802e0cc0b4 100644 --- a/src/main/cpp/velox-ref.txt +++ b/src/main/cpp/velox-ref.txt @@ -1 +1 @@ -01834e806060609c0557e7d66b5a5b63f0f4b469 +01834e806fc4cfa04cb47b5487a779544820d2b3 From 4d51f157ccba1329dfee024f900480094073943c Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Sat, 27 Jun 2026 12:09:01 +0800 Subject: [PATCH 05/10] Restore barrier checkpoint code after merge with gluten-0530 --- src/main/cpp/main/velox4j/jni/JniWrapper.cc | 34 +++++++++++++++++++ .../velox4j/query/StatefulQueryExecutor.cc | 6 +++- .../velox4j/query/StatefulQueryExecutor.h | 2 ++ src/main/cpp/velox-ref.txt | 2 +- .../zhztheplayer/velox4j/jni/JniApi.java | 5 +++ .../zhztheplayer/velox4j/jni/JniWrapper.java | 2 ++ .../velox4j/query/SerialTask.java | 5 +++ .../velox4j/stateful/StatefulElement.java | 8 +++++ .../velox4j/stateful/StatefulRecord.java | 5 +++ .../velox4j/stateful/StatefulWatermark.java | 5 +++ 10 files changed, 72 insertions(+), 2 deletions(-) diff --git a/src/main/cpp/main/velox4j/jni/JniWrapper.cc b/src/main/cpp/main/velox4j/jni/JniWrapper.cc index 70f3c2e91b..05433a42aa 100644 --- a/src/main/cpp/main/velox4j/jni/JniWrapper.cc +++ b/src/main/cpp/main/velox4j/jni/JniWrapper.cc @@ -180,6 +180,21 @@ jobject statefulTaskGet(JNIEnv* env, jobject javaThis, jlong itrId) { jobject result = env->NewObject(resultClass, constructor, id, watermark->timestamp()); + env->DeleteLocalRef(id); + env->DeleteLocalRef(resultClass); + return result; + } else if (element->isBarrier()) { + auto barrier = std::static_pointer_cast(element); + + jclass resultClass = env->FindClass( + "io/github/zhztheplayer/velox4j/stateful/StatefulBarrier"); + jmethodID constructor = + env->GetMethodID(resultClass, "", "(Ljava/lang/String;J)V"); + + jstring id = env->NewStringUTF(barrier->nodeId().c_str()); + jobject result = + env->NewObject(resultClass, constructor, id, barrier->checkpointId()); + env->DeleteLocalRef(id); env->DeleteLocalRef(resultClass); return result; @@ -351,6 +366,18 @@ void notifyCheckpointAborted( JNI_METHOD_END() } + +void injectBarrier( + JNIEnv* env, + jobject javaThis, + jlong itrId, + jlong checkpointId) { + JNI_METHOD_START + auto itr = ObjectStore::retrieve(itrId); + itr->injectBarrier(checkpointId); + JNI_METHOD_END() +} + jlong createExternalStreamFromDownIterator( JNIEnv* env, jobject javaThis, @@ -671,6 +698,13 @@ void JniWrapper::initialize(JNIEnv* env) { kTypeLong, kTypeLong, nullptr); + addNativeMethod( + "injectBarrier", + (void*)injectBarrier, + kTypeVoid, + kTypeLong, + kTypeLong, + nullptr); addNativeMethod( "createExternalStreamFromDownIterator", (void*)createExternalStreamFromDownIterator, diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc index ef179154a1..65e9f85c74 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc @@ -170,7 +170,6 @@ void StatefulSerialTask::snapshotState(long checkpointId) { std::vector StatefulSerialTask::snapshotSourceState() { return sourceStateSnapshots_; } -} std::vector StatefulSerialTask::notifyCheckpointComplete( long checkpointId) { @@ -233,4 +232,9 @@ std::unique_ptr StatefulQueryExecutor::execute() const { return std::make_unique(memoryManager_, query_); } + +void StatefulSerialTask::injectBarrier(long checkpointId) { + task_->injectBarrier(checkpointId); +} + } // namespace velox4j diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h index 2513c1aa30..b35b35f20d 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h @@ -65,6 +65,8 @@ class StatefulSerialTask : public UpIterator { void notifyCheckpointAborted(long checkpointId); + void injectBarrier(long checkpointId); + void addSplit( const facebook::velox::core::PlanNodeId& planNodeId, int32_t groupId, diff --git a/src/main/cpp/velox-ref.txt b/src/main/cpp/velox-ref.txt index b5fe1b8a0c..b98cd7da87 100644 --- a/src/main/cpp/velox-ref.txt +++ b/src/main/cpp/velox-ref.txt @@ -1 +1 @@ -dd42008e5a4d05ca25eb1827b235f6a17fb8e35a +1e25d21fb8b31f0040e9371156b05826d0f62900 diff --git a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java index 074b00bf01..8b76bcfe50 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java @@ -138,6 +138,11 @@ public void notifyCheckpointAborted(UpIterator itr, long checkpointId) { jni.notifyCheckpointAborted(itr.id(), checkpointId); } + // This method is for Flink + public void injectBarrier(UpIterator itr, long checkpointId) { + jni.injectBarrier(itr.id(), checkpointId); + } + public ExternalStream createExternalStreamFromDownIterator(DownIterator itr) { return new ExternalStreams.GenericExternalStream(jni.createExternalStreamFromDownIterator(itr)); } diff --git a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java index d9304c737c..08d135fa27 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java @@ -76,6 +76,8 @@ native void initializeState( native void notifyCheckpointAborted(long id, long checkpointId); + native void injectBarrier(long id, long checkpointId); + native long createBlockingQueue(); // For BaseVector / RowVector / SelectivityVector. diff --git a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java index 582bff34d4..e6fc8946eb 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java @@ -122,6 +122,11 @@ public void notifyCheckpointAborted(long checkpointId) { jniApi.notifyCheckpointAborted(this, checkpointId); } + // This method is for Flink + public void injectBarrier(long checkpointId) { + jniApi.injectBarrier(this, checkpointId); + } + @Override public void close() { StaticJniApi.get().releaseCppObject(this); diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java index 3b7f58ec3d..81a3663d2c 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java @@ -31,6 +31,10 @@ public String getNodeId() { public abstract boolean isRecord(); + public boolean isBarrier() { + return false; + } + public StatefulWatermark asWatermark() { return (StatefulWatermark) this; } @@ -39,5 +43,9 @@ public StatefulRecord asRecord() { return (StatefulRecord) this; } + public StatefulBarrier asBarrier() { + return (StatefulBarrier) this; + } + public void close() {} } diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java index 6b7775b8ad..67ed3dba70 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java @@ -60,6 +60,11 @@ public boolean isRecord() { return true; } + @Override + public boolean isBarrier() { + return false; + } + public boolean hasTimestamp() { return hasTimestamp; } diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java index 3ce092ba69..f7774c3320 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java @@ -37,4 +37,9 @@ public boolean isWatermark() { public boolean isRecord() { return false; } + + @Override + public boolean isBarrier() { + return false; + } } From 2af6d2feb7ebda3e1b973cd02c840dc89af8ce4e Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Sat, 27 Jun 2026 12:13:42 +0800 Subject: [PATCH 06/10] Fix snapshotState return type to match 0530 API after merge --- src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc index 65e9f85c74..a8d891b96b 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc @@ -163,8 +163,9 @@ void StatefulSerialTask::initializeState( } } -void StatefulSerialTask::snapshotState(long checkpointId) { +std::vector StatefulSerialTask::snapshotState(long checkpointId) { sourceStateSnapshots_ = task_->snapshotState(checkpointId); + return sourceStateSnapshots_; } std::vector StatefulSerialTask::snapshotSourceState() { From adf4dffbbc56cb10c63dab5fb7d360f0fc6dd87c Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Mon, 29 Jun 2026 12:06:07 +0800 Subject: [PATCH 07/10] Revert "Restore barrier checkpoint code after merge with gluten-0530" This reverts commit 4d51f157ccba1329dfee024f900480094073943c. --- src/main/cpp/main/velox4j/jni/JniWrapper.cc | 34 ------------------- .../velox4j/query/StatefulQueryExecutor.cc | 6 +--- .../velox4j/query/StatefulQueryExecutor.h | 2 -- src/main/cpp/velox-ref.txt | 2 +- .../zhztheplayer/velox4j/jni/JniApi.java | 5 --- .../zhztheplayer/velox4j/jni/JniWrapper.java | 2 -- .../velox4j/query/SerialTask.java | 5 --- .../velox4j/stateful/StatefulElement.java | 8 ----- .../velox4j/stateful/StatefulRecord.java | 5 --- .../velox4j/stateful/StatefulWatermark.java | 5 --- 10 files changed, 2 insertions(+), 72 deletions(-) diff --git a/src/main/cpp/main/velox4j/jni/JniWrapper.cc b/src/main/cpp/main/velox4j/jni/JniWrapper.cc index 05433a42aa..70f3c2e91b 100644 --- a/src/main/cpp/main/velox4j/jni/JniWrapper.cc +++ b/src/main/cpp/main/velox4j/jni/JniWrapper.cc @@ -180,21 +180,6 @@ jobject statefulTaskGet(JNIEnv* env, jobject javaThis, jlong itrId) { jobject result = env->NewObject(resultClass, constructor, id, watermark->timestamp()); - env->DeleteLocalRef(id); - env->DeleteLocalRef(resultClass); - return result; - } else if (element->isBarrier()) { - auto barrier = std::static_pointer_cast(element); - - jclass resultClass = env->FindClass( - "io/github/zhztheplayer/velox4j/stateful/StatefulBarrier"); - jmethodID constructor = - env->GetMethodID(resultClass, "", "(Ljava/lang/String;J)V"); - - jstring id = env->NewStringUTF(barrier->nodeId().c_str()); - jobject result = - env->NewObject(resultClass, constructor, id, barrier->checkpointId()); - env->DeleteLocalRef(id); env->DeleteLocalRef(resultClass); return result; @@ -366,18 +351,6 @@ void notifyCheckpointAborted( JNI_METHOD_END() } - -void injectBarrier( - JNIEnv* env, - jobject javaThis, - jlong itrId, - jlong checkpointId) { - JNI_METHOD_START - auto itr = ObjectStore::retrieve(itrId); - itr->injectBarrier(checkpointId); - JNI_METHOD_END() -} - jlong createExternalStreamFromDownIterator( JNIEnv* env, jobject javaThis, @@ -698,13 +671,6 @@ void JniWrapper::initialize(JNIEnv* env) { kTypeLong, kTypeLong, nullptr); - addNativeMethod( - "injectBarrier", - (void*)injectBarrier, - kTypeVoid, - kTypeLong, - kTypeLong, - nullptr); addNativeMethod( "createExternalStreamFromDownIterator", (void*)createExternalStreamFromDownIterator, diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc index a8d891b96b..0738c01d96 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc @@ -171,6 +171,7 @@ std::vector StatefulSerialTask::snapshotState(long checkpointId) { std::vector StatefulSerialTask::snapshotSourceState() { return sourceStateSnapshots_; } +} std::vector StatefulSerialTask::notifyCheckpointComplete( long checkpointId) { @@ -233,9 +234,4 @@ std::unique_ptr StatefulQueryExecutor::execute() const { return std::make_unique(memoryManager_, query_); } - -void StatefulSerialTask::injectBarrier(long checkpointId) { - task_->injectBarrier(checkpointId); -} - } // namespace velox4j diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h index b35b35f20d..2513c1aa30 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.h @@ -65,8 +65,6 @@ class StatefulSerialTask : public UpIterator { void notifyCheckpointAborted(long checkpointId); - void injectBarrier(long checkpointId); - void addSplit( const facebook::velox::core::PlanNodeId& planNodeId, int32_t groupId, diff --git a/src/main/cpp/velox-ref.txt b/src/main/cpp/velox-ref.txt index b98cd7da87..b5fe1b8a0c 100644 --- a/src/main/cpp/velox-ref.txt +++ b/src/main/cpp/velox-ref.txt @@ -1 +1 @@ -1e25d21fb8b31f0040e9371156b05826d0f62900 +dd42008e5a4d05ca25eb1827b235f6a17fb8e35a diff --git a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java index 8b76bcfe50..074b00bf01 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniApi.java @@ -138,11 +138,6 @@ public void notifyCheckpointAborted(UpIterator itr, long checkpointId) { jni.notifyCheckpointAborted(itr.id(), checkpointId); } - // This method is for Flink - public void injectBarrier(UpIterator itr, long checkpointId) { - jni.injectBarrier(itr.id(), checkpointId); - } - public ExternalStream createExternalStreamFromDownIterator(DownIterator itr) { return new ExternalStreams.GenericExternalStream(jni.createExternalStreamFromDownIterator(itr)); } diff --git a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java index 08d135fa27..d9304c737c 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/jni/JniWrapper.java @@ -76,8 +76,6 @@ native void initializeState( native void notifyCheckpointAborted(long id, long checkpointId); - native void injectBarrier(long id, long checkpointId); - native long createBlockingQueue(); // For BaseVector / RowVector / SelectivityVector. diff --git a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java index e6fc8946eb..582bff34d4 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/query/SerialTask.java @@ -122,11 +122,6 @@ public void notifyCheckpointAborted(long checkpointId) { jniApi.notifyCheckpointAborted(this, checkpointId); } - // This method is for Flink - public void injectBarrier(long checkpointId) { - jniApi.injectBarrier(this, checkpointId); - } - @Override public void close() { StaticJniApi.get().releaseCppObject(this); diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java index 81a3663d2c..3b7f58ec3d 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulElement.java @@ -31,10 +31,6 @@ public String getNodeId() { public abstract boolean isRecord(); - public boolean isBarrier() { - return false; - } - public StatefulWatermark asWatermark() { return (StatefulWatermark) this; } @@ -43,9 +39,5 @@ public StatefulRecord asRecord() { return (StatefulRecord) this; } - public StatefulBarrier asBarrier() { - return (StatefulBarrier) this; - } - public void close() {} } diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java index 67ed3dba70..6b7775b8ad 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulRecord.java @@ -60,11 +60,6 @@ public boolean isRecord() { return true; } - @Override - public boolean isBarrier() { - return false; - } - public boolean hasTimestamp() { return hasTimestamp; } diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java index f7774c3320..3ce092ba69 100644 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java +++ b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulWatermark.java @@ -37,9 +37,4 @@ public boolean isWatermark() { public boolean isRecord() { return false; } - - @Override - public boolean isBarrier() { - return false; - } } From 6e7b99ae72d3dc71f65be6a78b03b86d0c80a1f1 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Mon, 29 Jun 2026 12:06:07 +0800 Subject: [PATCH 08/10] Revert "Fix snapshotState return type to match 0530 API after merge" This reverts commit 2af6d2feb7ebda3e1b973cd02c840dc89af8ce4e. --- src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc index 0738c01d96..ef179154a1 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc @@ -163,9 +163,8 @@ void StatefulSerialTask::initializeState( } } -std::vector StatefulSerialTask::snapshotState(long checkpointId) { +void StatefulSerialTask::snapshotState(long checkpointId) { sourceStateSnapshots_ = task_->snapshotState(checkpointId); - return sourceStateSnapshots_; } std::vector StatefulSerialTask::snapshotSourceState() { From 16513d9ded4d6a98a304c2d27e5c2a782e5c1896 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Mon, 29 Jun 2026 14:38:28 +0800 Subject: [PATCH 09/10] revert barrier --- .../velox4j/stateful/StatefulBarrier.java | 44 --------------- .../velox4j/stateful/StatefulBarrierTest.java | 55 ------------------- 2 files changed, 99 deletions(-) delete mode 100644 src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrier.java delete mode 100644 src/test/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrierTest.java diff --git a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrier.java b/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrier.java deleted file mode 100644 index 065a1b532e..0000000000 --- a/src/main/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrier.java +++ /dev/null @@ -1,44 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one or more -* contributor license agreements. See the NOTICE file distributed with -* this work for additional information regarding copyright ownership. -* The ASF licenses this file to You under the Apache License, Version 2.0 -* (the "License"); you may not use this file except in compliance with -* the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, software -* distributed under the License is distributed on an "AS IS" BASIS, -* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -* See the License for the specific language governing permissions and -* limitations under the License. -*/ -package io.github.zhztheplayer.velox4j.stateful; - -public class StatefulBarrier extends StatefulElement { - private final long checkpointId; - - public StatefulBarrier(String nodeId, long checkpointId) { - super(nodeId); - this.checkpointId = checkpointId; - } - - public long getCheckpointId() { - return checkpointId; - } - - @Override - public boolean isWatermark() { - return false; - } - - @Override - public boolean isRecord() { - return false; - } - - public boolean isBarrier() { - return true; - } -} diff --git a/src/test/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrierTest.java b/src/test/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrierTest.java deleted file mode 100644 index 6221235e91..0000000000 --- a/src/test/java/io/github/zhztheplayer/velox4j/stateful/StatefulBarrierTest.java +++ /dev/null @@ -1,55 +0,0 @@ -/* -* Licensed to the Apache Software Foundation (ASF) under one or more -* contributor license agreements. See the NOTICE file distributed with -* this work for additional information regarding copyright ownership. -* The ASF licenses this file to You under the Apache License, Version 2.0 -* (the "License"); you may not use this file except in compliance with -* the License. You may obtain a copy of the License at -* -* http://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, software -* distributed under the License is distributed on an "AS IS" BASIS, -* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -* See the License for the specific language governing permissions and -* limitations under the License. -*/ -package io.github.zhztheplayer.velox4j.stateful; - -import org.junit.Assert; -import org.junit.Test; - -public class StatefulBarrierTest { - - @Test - public void testBarrierProperties() { - StatefulBarrier barrier = new StatefulBarrier("node-1", 42L); - Assert.assertEquals("node-1", barrier.getNodeId()); - Assert.assertEquals(42L, barrier.getCheckpointId()); - Assert.assertTrue(barrier.isBarrier()); - Assert.assertFalse(barrier.isWatermark()); - Assert.assertFalse(barrier.isRecord()); - } - - @Test - public void testAsBarrier() { - StatefulElement element = new StatefulBarrier("node-2", 99L); - Assert.assertTrue(element.isBarrier()); - StatefulBarrier barrier = element.asBarrier(); - Assert.assertEquals(99L, barrier.getCheckpointId()); - } - - @Test - public void testRecordIsNotBarrier() { - StatefulElement record = new StatefulRecord("node-1", 0L, -1L, false, -1); - Assert.assertFalse(record.isBarrier()); - Assert.assertTrue(record.isRecord()); - } - - @Test - public void testWatermarkIsNotBarrier() { - StatefulElement watermark = new StatefulWatermark("node-1", 1000L); - Assert.assertFalse(watermark.isBarrier()); - Assert.assertTrue(watermark.isWatermark()); - } -} From e7d2a2fdfa33e7a534b1863300e42b4e237154c6 Mon Sep 17 00:00:00 2001 From: zhangzhibiao Date: Mon, 29 Jun 2026 14:51:35 +0800 Subject: [PATCH 10/10] Fix StatefulQueryExecutor: snapshotState return type and stray namespace brace --- src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc index ef179154a1..2e9e179960 100644 --- a/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc +++ b/src/main/cpp/main/velox4j/query/StatefulQueryExecutor.cc @@ -163,14 +163,14 @@ void StatefulSerialTask::initializeState( } } -void StatefulSerialTask::snapshotState(long checkpointId) { +std::vector StatefulSerialTask::snapshotState(long checkpointId) { sourceStateSnapshots_ = task_->snapshotState(checkpointId); + return sourceStateSnapshots_; } std::vector StatefulSerialTask::snapshotSourceState() { return sourceStateSnapshots_; } -} std::vector StatefulSerialTask::notifyCheckpointComplete( long checkpointId) {