snapshot id + 1

alibaba · Oct 22, 2024 · 59fec6a · 59fec6a
1 parent 3b6adce
commit 59fec6a
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 15 deletions.
diff --git a/interactive_engine/executor/store/mcsr/src/graph_partitioner.rs b/interactive_engine/executor/store/mcsr/src/graph_partitioner.rs
@@ -214,19 +214,16 @@ impl<G: FromStr + Send + Sync + IndexType + Eq> GraphPartitioner<G> {
                 {
                     let input_path = vertex_file
                         .as_os_str()
-                        .clone()
                         .to_str()
                         .unwrap();
                     let input_dir_path = self
                         .input_dir
                         .as_os_str()
-                        .clone()
                         .to_str()
                         .unwrap();
                     let output_path = if let Some(pos) = input_path.find(input_dir_path) {
                         self.partition_dir.join(
                             input_path
-                                .clone()
                                 .split_at(pos + input_dir_path.len() + 1)
                                 .1,
                         )
@@ -260,21 +257,18 @@ impl<G: FromStr + Send + Sync + IndexType + Eq> GraphPartitioner<G> {
                 {
                     let input_path = vertex_file
                         .as_os_str()
-                        .clone()
                         .to_str()
                         .unwrap();
                     let input_dir_path = self
                         .input_dir
                         .as_os_str()
-                        .clone()
                         .to_str()
                         .unwrap();
                     let gz_loc = input_path.find(".gz").unwrap();
                     let input_path = input_path.split_at(gz_loc).0;
                     let output_path = if let Some(pos) = input_path.find(input_dir_path) {
                         self.partition_dir.join(
                             input_path
-                                .clone()
                                 .split_at(pos + input_dir_path.len() + 1)
                                 .1,
                         )
@@ -327,18 +321,16 @@ impl<G: FromStr + Send + Sync + IndexType + Eq> GraphPartitioner<G> {
                                 .unwrap()
                                 .ends_with(".csv")
                             {
-                                info!("{}", edge_file.as_os_str().clone().to_str().unwrap());
-                                let input_path = edge_file.as_os_str().clone().to_str().unwrap();
+                                info!("{}", edge_file.as_os_str().to_str().unwrap());
+                                let input_path = edge_file.as_os_str().to_str().unwrap();
                                 let input_dir_path = self
                                     .input_dir
                                     .as_os_str()
-                                    .clone()
                                     .to_str()
                                     .unwrap();
                                 let output_path = if let Some(pos) = input_path.find(input_dir_path) {
                                     self.partition_dir.join(
                                         input_path
-                                            .clone()
                                             .split_at(pos + input_dir_path.len() + 1)
                                             .1,
                                     )
@@ -373,19 +365,17 @@ impl<G: FromStr + Send + Sync + IndexType + Eq> GraphPartitioner<G> {
                                 .unwrap()
                                 .ends_with(".csv.gz")
                             {
-                                let input_path = edge_file.as_os_str().clone().to_str().unwrap();
+                                let input_path = edge_file.as_os_str().to_str().unwrap();
                                 let input_dir_path = self
                                     .input_dir
                                     .as_os_str()
-                                    .clone()
                                     .to_str()
                                     .unwrap();
                                 let gz_loc = input_path.find(".gz").unwrap();
                                 let input_path = input_path.split_at(gz_loc).0;
                                 let output_path = if let Some(pos) = input_path.find(input_dir_path) {
                                     self.partition_dir.join(
                                         input_path
-                                            .clone()
                                             .split_at(pos + input_dir_path.len() + 1)
                                             .1,
                                     )

diff --git a/...groot-module/src/main/java/com/alibaba/graphscope/groot/frontend/write/KafkaAppender.java b/...groot-module/src/main/java/com/alibaba/graphscope/groot/frontend/write/KafkaAppender.java
@@ -259,7 +259,7 @@ public List<Long> replayDMLRecordsFrom(long offset, long timestamp) throws IOExc
                         logWriter.append(storeId, new LogEntry(batchSnapshotId, batch));
                         replayCount++;
                     }
-                    ids.add(batchSnapshotId);
+                    ids.add(batchSnapshotId + 1);
                 }
             }
         }

diff --git a/..._engine/groot-module/src/main/java/com/alibaba/graphscope/groot/store/KafkaProcessor.java b/..._engine/groot-module/src/main/java/com/alibaba/graphscope/groot/store/KafkaProcessor.java
@@ -393,6 +393,6 @@ public List<Long> replayDMLRecordsFrom(long offset, long timestamp) throws IOExc
             replayInProgress.set(false);
         }
         logger.info("replay DML records finished. total replayed [{}] records", replayCount);
-        return List.of(batchSnapshotId);
+        return List.of(batchSnapshotId + 1);
     }
 }