fix(interactive): Fix several sparodic bugs in groot (#4595)

siyuan0322 · web-flow · commit e5128200f4a9 · 2025-06-05T19:00:02.000+08:00
and bump up rustc version to 1.87.0
Signed-off-by: siyuan0322 &lt;siyuanzhang.zsy@alibaba-inc.com&gt;
diff --git a/.github/workflows/gaia.yml b/.github/workflows/gaia.yml
@@ -55,7 +55,7 @@ jobs:
     - name: Install Rust
       uses: actions-rs/toolchain@v1
       with:
-        toolchain: 1.81.0
+        toolchain: 1.87.0
         override: true
 
     - name: Rust Format Check
diff --git a/.github/workflows/gss.yml b/.github/workflows/gss.yml
@@ -79,8 +79,8 @@ jobs:
         . ${HOME}/.graphscope_env
         export SCCACHE_DIR=~/.cache/sccache
         export RUSTC_WRAPPER=/usr/local/bin/sccache
-        rustup toolchain install 1.81.0
-        rustup default 1.81.0
+        rustup toolchain install 1.87.0
+        rustup default 1.87.0
         cd ${GITHUB_WORKSPACE}/interactive_engine
         mvn clean install -P groot -Drust.compile.mode=debug -DskipTests --quiet
         mvn clean install -Pgroot-data-load --quiet
diff --git a/.github/workflows/interactive.yml b/.github/workflows/interactive.yml
@@ -129,6 +129,7 @@ jobs:
         . /home/graphscope/.cargo/env
         which cargo
 
+        rustup toolchain install 1.87.0 && rustup default 1.87.0
         # build compiler
         cd ${GIE_HOME}/
         mvn clean install -Pexperimental -DskipTests -q
diff --git a/.github/workflows/k8s-ci.yml b/.github/workflows/k8s-ci.yml
@@ -285,7 +285,7 @@ jobs:
     - name: Install Rust
       uses: actions-rs/toolchain@v1
       with:
-        toolchain: 1.81.0
+        toolchain: 1.87.0
         override: true
 
     - name: Build Artifact
@@ -649,7 +649,7 @@ jobs:
       - name: Install Rust
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.81.0
+          toolchain: 1.87.0
           override: true
 
       - name: Build GIE Experimental Artifacts
diff --git a/interactive_engine/executor/store/groot/Cargo.toml b/interactive_engine/executor/store/groot/Cargo.toml
@@ -22,8 +22,7 @@ grpcio = "0.10"
 grpcio-sys = { version = "0.10", features = ["openssl"] }
 # deactivation of bzip2 due to https://github.com/rust-rocksdb/rust-rocksdb/issues/609
 # deactivation of zstd due to the 'hidden symbol "ZSTD_maxCLevel" is referenced by DSO' error
-rocksdb = { version = "0.21.0", features = ["snappy", "lz4", "zlib"], default-features = false }
-#rocksdb = { git = "https://github.com/siyuan0322/rust-rocksdb.git", rev = "c44ea2b", features = ["snappy", "lz4", "zlib"], default-features = false }
+rocksdb = { version = "0.21.0", features = ["snappy", "lz4", "zlib", "multi-threaded-cf"], default-features = false }
 dyn_type = { path = "../../common/dyn_type" }
 rustversion = "1.0"
 mimalloc-rust = {version = "0.2.1", optional = true}
diff --git a/interactive_engine/executor/store/groot/src/db/graph/store.rs b/interactive_engine/executor/store/groot/src/db/graph/store.rs
@@ -849,7 +849,9 @@ impl GraphStore {
 
     fn check_si_guard(&self, si: SnapshotId) -> GraphResult<()> {
         let guard = self.si_guard.load(Ordering::Relaxed) as SnapshotId;
-        if si < guard {
+        // allow for a short duration inconsistent, due to different frontend may have minor
+        // difference in timing
+        if guard - si > 10 {
             let msg = format!("si#{} is less than current si_guard#{}", si, guard);
             let err = gen_graph_err!(ErrorCode::INVALID_OPERATION, msg);
             return Err(err);
diff --git a/interactive_engine/executor/store/groot/src/db/graph/tests/graph.rs b/interactive_engine/executor/store/groot/src/db/graph/tests/graph.rs
@@ -19,15 +19,15 @@ pub fn test_si_guard<G: MultiVersionGraph>(graph: G) {
         .unwrap();
     schema_version += 1;
     assert!(graph
-        .create_vertex_type(11, schema_version, 4, &types::create_test_type_def(4), schema_version)
+        .create_vertex_type(1, schema_version, 4, &types::create_test_type_def(4), schema_version)
         .is_err());
     schema_version += 1;
     graph
         .drop_vertex_type(15, schema_version, 1)
         .unwrap();
     schema_version += 1;
     assert!(graph
-        .create_vertex_type(13, schema_version, 5, &types::create_test_type_def(5), schema_version)
+        .create_vertex_type(3, schema_version, 5, &types::create_test_type_def(5), schema_version)
         .is_err());
     let properties: HashMap<PropertyId, Value> = HashMap::new();
     graph
@@ -38,21 +38,21 @@ pub fn test_si_guard<G: MultiVersionGraph>(graph: G) {
         .unwrap();
     schema_version += 1;
     assert!(graph
-        .create_vertex_type(16, schema_version, 5, &types::create_test_type_def(5), schema_version)
+        .create_vertex_type(6, schema_version, 5, &types::create_test_type_def(5), schema_version)
         .is_err());
     schema_version += 1;
     assert!(graph
-        .drop_vertex_type(16, schema_version, 2)
+        .drop_vertex_type(6, schema_version, 2)
         .is_err());
     schema_version += 1;
     assert!(graph
-        .insert_overwrite_vertex(16, 1, 2, &properties)
+        .insert_overwrite_vertex(6, 1, 2, &properties)
         .is_err());
     schema_version += 1;
-    assert!(graph.delete_vertex(16, 1, 2).is_err());
+    assert!(graph.delete_vertex(6, 1, 2).is_err());
     schema_version += 1;
     assert!(graph
-        .insert_update_vertex(16, 1, 2, &properties)
+        .insert_update_vertex(6, 1, 2, &properties)
         .is_err());
     schema_version += 1;
 
@@ -66,7 +66,7 @@ pub fn test_si_guard<G: MultiVersionGraph>(graph: G) {
         .unwrap();
     schema_version += 1;
     assert!(graph
-        .insert_overwrite_vertex(16, 1, 2, &properties)
+        .insert_overwrite_vertex(6, 1, 2, &properties)
         .is_err());
     graph
         .insert_update_edge(20, EdgeId::new(1, 2, 3), &edge_type, true, &properties)
@@ -78,24 +78,24 @@ pub fn test_si_guard<G: MultiVersionGraph>(graph: G) {
         .insert_overwrite_edge(21, EdgeId::new(1, 2, 3), &edge_type, true, &properties)
         .unwrap();
     assert!(graph
-        .insert_update_edge(20, EdgeId::new(1, 2, 3), &edge_type, true, &properties)
+        .insert_update_edge(10, EdgeId::new(1, 2, 3), &edge_type, true, &properties)
         .is_err());
     assert!(graph
-        .drop_edge_type(19, schema_version, 10)
+        .drop_edge_type(9, schema_version, 10)
         .is_err());
     schema_version += 1;
     assert!(graph
-        .remove_edge_kind(19, schema_version, &edge_type)
+        .remove_edge_kind(9, schema_version, &edge_type)
         .is_err());
     schema_version += 1;
     assert!(graph
-        .delete_edge(19, EdgeId::new(1, 2, 3), &edge_type, true)
+        .delete_edge(9, EdgeId::new(1, 2, 3), &edge_type, true)
         .is_err());
     assert!(graph
-        .create_edge_type(19, schema_version, 20, &types::create_test_type_def(2))
+        .create_edge_type(9, schema_version, 20, &types::create_test_type_def(2))
         .is_err());
     schema_version += 1;
     assert!(graph
-        .add_edge_kind(19, schema_version, &edge_type, schema_version)
+        .add_edge_kind(9, schema_version, &edge_type, schema_version)
         .is_err());
 }
diff --git a/interactive_engine/executor/store/groot/src/db/storage/rocksdb.rs b/interactive_engine/executor/store/groot/src/db/storage/rocksdb.rs
@@ -7,7 +7,7 @@ use std::time::Duration;
 use ::rocksdb::backup::{BackupEngine, BackupEngineOptions, RestoreOptions};
 use ::rocksdb::{DBRawIterator, Env, IngestExternalFileOptions, Options, ReadOptions, DB};
 use crossbeam_epoch::{self as epoch, Atomic, Owned};
-use rocksdb::WriteBatch;
+use rocksdb::{WriteBatch, WriteOptions};
 
 use super::{StorageIter, StorageRes};
 use crate::db::api::*;
@@ -85,7 +85,12 @@ impl RocksDB {
 
         let prev = self.db.swap(cur, Ordering::Release, guard);
         unsafe {
-            drop(prev.into_owned());
+            let prev = prev.into_owned();
+            prev.cancel_all_background_work(true);
+            std::thread::spawn(move || {
+                std::thread::sleep(Duration::from_secs(120));
+                drop(prev);
+            });
         }
         info!("RocksDB replaced");
     }
@@ -109,7 +114,9 @@ impl RocksDB {
             return Ok(());
         }
         let db = self.get_db()?;
-        db.put(key, val).map_err(|e| {
+        let mut opt = WriteOptions::default();
+        opt.set_sync(false);
+        db.put_opt(key, val, &opt).map_err(|e| {
             let msg = format!("rocksdb.put failed because {}", e.into_string());
             gen_graph_err!(ErrorCode::EXTERNAL_STORAGE_ERROR, msg)
         })
@@ -381,8 +388,11 @@ fn init_options(options: &HashMap<String, String>) -> Options {
     // https://github.com/facebook/rocksdb/wiki/Basic-Operations#non-sync-writes
     opts.set_use_fsync(true);
     opts.set_max_write_buffer_number(4);
-
+    opts.set_allow_concurrent_memtable_write(true);
+    opts.set_enable_write_thread_adaptive_yield(true);
     opts.set_bytes_per_sync(1048576);
+    opts.set_enable_pipelined_write(true);
+    opts.set_unordered_write(false);
 
     if let Some(conf_str) = options.get("store.rocksdb.disable.auto.compactions") {
         let val = conf_str.parse().unwrap();
diff --git a/interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/frontend/write/KafkaAppender.java b/interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/frontend/write/KafkaAppender.java
@@ -19,6 +19,7 @@
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.metrics.Meter;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -248,9 +249,9 @@ public List<Long> replayDMLRecordsFrom(long offset, long timestamp) throws IOExc
                 try (LogReader logReader =
                         this.logService.createReader(storeId, offset, timestamp)) {
                     long batchSnapshotId = ingestSnapshotId.get();
-                    ReadLogEntry readLogEntry;
-                    while (!shouldStop && (readLogEntry = logReader.readNext()) != null) {
-                        LogEntry logEntry = readLogEntry.getLogEntry();
+                    ConsumerRecord<LogEntry, LogEntry> readLogEntry;
+                    while (!shouldStop && (readLogEntry = logReader.readNextRecord()) != null) {
+                        LogEntry logEntry = readLogEntry.value();
                         OperationBatch batch =
                                 Utils.extractOperations(logEntry.getOperationBatch(), types);
                         if (batch.getOperationCount() == 0) {
diff --git a/interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/store/KafkaProcessor.java b/interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/store/KafkaProcessor.java
@@ -372,16 +372,15 @@ public List<Long> replayDMLRecordsFrom(long offset, long timestamp) throws IOExc
         long batchSnapshotId;
         int replayCount = 0;
         try (LogReader logReader = this.logService.createReader(storeId, offset, timestamp)) {
-            ReadLogEntry readLogEntry;
+            ConsumerRecord<LogEntry, LogEntry> record;
             batchSnapshotId = latestSnapshotId.get();
-            while (!shouldStop && (readLogEntry = logReader.readNext()) != null) {
-                LogEntry logEntry = readLogEntry.getLogEntry();
+            while (!shouldStop && (record = logReader.readNextRecord()) != null) {
+                LogEntry logEntry = record.value();
                 OperationBatch batch = Utils.extractOperations(logEntry.getOperationBatch(), types);
                 if (batch.getOperationCount() == 0) {
                     continue;
                 }
-                ReadLogEntry entry =
-                        new ReadLogEntry(readLogEntry.getOffset(), batchSnapshotId, batch);
+                ReadLogEntry entry = new ReadLogEntry(record.offset(), batchSnapshotId, batch);
                 replayQueue.put(entry);
                 replayCount++;
                 if (replayCount % 10000 == 0) {
diff --git a/interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogReader.java b/interactive_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogReader.java
@@ -24,6 +24,7 @@
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,6 +44,8 @@ public class KafkaLogReader implements LogReader {
     private final long latest;
     private long nextReadOffset;
 
+    private TopicPartition partition;
+
     public KafkaLogReader(
             String servers,
             AdminClient client,
@@ -53,8 +56,9 @@ public KafkaLogReader(
             throws IOException {
         Map<String, Object> kafkaConfigs = new HashMap<>();
         kafkaConfigs.put("bootstrap.servers", servers);
+        // kafkaConfigs.put("max.poll.records", 100);
 
-        TopicPartition partition = new TopicPartition(topicName, partitionId);
+        partition = new TopicPartition(topicName, partitionId);
 
         long earliest = getOffset(client, partition, OffsetSpec.earliest());
         latest = getOffset(client, partition, OffsetSpec.latest());
@@ -91,9 +95,10 @@ private long getOffset(AdminClient client, TopicPartition partition, OffsetSpec
         }
     }
 
+    // Deprecated
     @Override
     public ReadLogEntry readNext() {
-        if (nextReadOffset == latest) {
+        if (nextReadOffset >= latest) {
             return null;
         }
         while (iterator == null || !iterator.hasNext()) {
@@ -113,17 +118,24 @@ public ReadLogEntry readNext() {
 
     @Override
     public ConsumerRecord<LogEntry, LogEntry> readNextRecord() {
-        if (nextReadOffset == latest) {
+        if (nextReadOffset >= latest) {
             return null;
         }
         while (iterator == null || !iterator.hasNext()) {
-            ConsumerRecords<LogEntry, LogEntry> consumerRecords =
-                    consumer.poll(Duration.ofMillis(100L));
-            if (consumerRecords == null || consumerRecords.isEmpty()) {
-                logger.info("polled nothing from Kafka. nextReadOffset is [{}]", nextReadOffset);
-                continue;
+            try {
+                ConsumerRecords<LogEntry, LogEntry> consumerRecords =
+                        consumer.poll(Duration.ofMillis(100L));
+                if (consumerRecords == null || consumerRecords.isEmpty()) {
+                    logger.info(
+                            "polled nothing from Kafka. nextReadOffset is [{}]", nextReadOffset);
+                    continue;
+                }
+                iterator = consumerRecords.iterator();
+            } catch (KafkaException ex) {
+                logger.error("poll failed, error: ", ex);
+                long cur = consumer.position(partition);
+                consumer.seek(partition, cur + 1);
             }
-            iterator = consumerRecords.iterator();
         }
         ConsumerRecord<LogEntry, LogEntry> record = iterator.next();
         nextReadOffset = record.offset() + 1;
diff --git a/interactive_engine/groot-server/src/test/java/com/alibaba/graphscope/groot/tests/common/wal/kafka/KafkaWalTest.java b/interactive_engine/groot-server/src/test/java/com/alibaba/graphscope/groot/tests/common/wal/kafka/KafkaWalTest.java
@@ -24,10 +24,10 @@
 import com.alibaba.graphscope.groot.wal.LogReader;
 import com.alibaba.graphscope.groot.wal.LogService;
 import com.alibaba.graphscope.groot.wal.LogWriter;
-import com.alibaba.graphscope.groot.wal.ReadLogEntry;
 import com.alibaba.graphscope.groot.wal.kafka.KafkaLogService;
 import com.salesforce.kafka.test.junit5.SharedKafkaTestResource;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
@@ -94,14 +94,14 @@ void testLogService() throws IOException {
         assertEquals(writer.append(queueId, logEntry), 0);
 
         LogReader reader = logService.createReader(queueId, 0);
-        ReadLogEntry readLogEntry = reader.readNext();
+        ConsumerRecord<LogEntry, LogEntry> record = reader.readNextRecord();
         reader.close();
 
         assertAll(
-                () -> assertEquals(readLogEntry.getOffset(), 0),
-                () -> assertEquals(readLogEntry.getLogEntry().getSnapshotId(), snapshotId));
+                () -> assertEquals(record.offset(), 0),
+                () -> assertEquals(record.value().getSnapshotId(), snapshotId));
 
-        OperationBatch operationBatch = readLogEntry.getLogEntry().getOperationBatch();
+        OperationBatch operationBatch = record.value().getOperationBatch();
         assertEquals(operationBatch.getOperationCount(), 1);
         assertEquals(operationBatch.getOperationBlob(0), OperationBlob.MARKER_OPERATION_BLOB);
 
@@ -110,7 +110,7 @@ void testLogService() throws IOException {
         assertEquals(writer.append(queueId, logEntry), 3);
 
         LogReader readerTail = logService.createReader(queueId, 4);
-        assertNull(readerTail.readNext());
+        assertNull(readerTail.readNextRecord());
         readerTail.close();
 
         assertThrows(IllegalArgumentException.class, () -> logService.createReader(queueId, 5));
diff --git a/k8s/dockerfiles/graphscope-store.Dockerfile b/k8s/dockerfiles/graphscope-store.Dockerfile
@@ -14,7 +14,7 @@ COPY --chown=graphscope:graphscope . /home/graphscope/graphscope
 COPY --chown=graphscope:graphscope ./interactive_engine/assembly/src/conf/maven.settings.xml /home/graphscope/.m2/settings.xml
 
 USER graphscope
-RUN rustup toolchain install 1.81.0 && rustup default 1.81.0
+RUN rustup toolchain install 1.87.0 && rustup default 1.87.0
 
 RUN cd /home/graphscope/graphscope \
     && . ~/.graphscope_env \
diff --git a/k8s/dockerfiles/interactive.Dockerfile b/k8s/dockerfiles/interactive.Dockerfile
@@ -18,7 +18,7 @@ RUN cd /home/graphscope/GraphScope/ && \
     else \
         mkdir /home/graphscope/install; \
         . /home/graphscope/.graphscope_env; \
-        rustup toolchain install 1.81.0 && rustup default 1.81.0; \
+        rustup toolchain install 1.87.0 && rustup default 1.87.0; \
         make interactive-install BUILD_TYPE="$profile" INSTALL_PREFIX=/home/graphscope/install; \
     fi
 
diff --git a/k8s/internal/Makefile b/k8s/internal/Makefile
@@ -110,7 +110,7 @@ graphscope-manylinux2014-py3-nodocker:
 	sudo sed -i 's|#baseurl=http://mirror.centos.org|baseurl=http://vault.centos.org|g' /etc/yum.repos.d/CentOS-* && \
 	sudo yum install java-11-openjdk-devel -y && \
 	sudo yum remove java-1.8.0-openjdk-devel java-1.8.0-openjdk java-1.8.0-openjdk-headless -y && \
-	rustup toolchain install 1.81.0 && rustup default 1.81.0 && \
+	rustup toolchain install 1.87.0 && rustup default 1.87.0 && \
 	cd $(WORKING_DIR)/../.. && \
 		if [[ "${PLATFORM}" == "aarch64" ]]; then \
 			export AUDITWHEEL_PLAT=manylinux2014_${PLATFORM}; \
diff --git a/python/graphscope/gsctl/scripts/install_deps.sh b/python/graphscope/gsctl/scripts/install_deps.sh
@@ -899,8 +899,8 @@ install_interactive_dependencies() {
   if ! command -v rustup &>/dev/null; then
     curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
     source $HOME/.cargo/env
-    rustup install 1.81.0
-    rustup default 1.81.0
+    rustup install 1.87.0
+    rustup default 1.87.0
     rustc --version
   fi
   # opentelemetry
