fix(all): version 0.8

builder/model/src/main/java/com/antgroup/openspg/builder/model/record/SubGraphRecord.java

@@ -22,8 +22,10 @@
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
+import lombok.Setter;
 
 @Getter
+@Setter
 @AllArgsConstructor
 @EqualsAndHashCode(callSuper = false)
 public class SubGraphRecord extends BaseRecord {

builder/runner/local/src/main/java/com/antgroup/openspg/builder/runner/local/physical/sink/impl/Neo4jSinkWriter.java

@@ -29,19 +29,14 @@
 import com.antgroup.openspg.cloudext.interfaces.graphstore.model.lpg.record.LPGPropertyRecord;
 import com.antgroup.openspg.cloudext.interfaces.graphstore.model.lpg.record.VertexRecord;
 import com.antgroup.openspg.cloudext.interfaces.graphstore.model.lpg.schema.EdgeTypeName;
+import com.antgroup.openspg.cloudext.interfaces.graphstore.util.TypeChecker;
 import com.antgroup.openspg.server.common.model.project.Project;
 import com.google.common.collect.Lists;
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.RejectedExecutionHandler;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -188,7 +183,11 @@ private void writeNode(SubGraphRecord.Node node) {
       List<VertexRecord> vertexRecords = Lists.newArrayList();
       List<LPGPropertyRecord> properties = Lists.newArrayList();
       for (Map.Entry<String, Object> entry : node.getProperties().entrySet()) {
-        properties.add(new LPGPropertyRecord(entry.getKey(), entry.getValue()));
+        Object entryValue = entry.getValue();
+        if (!TypeChecker.isArrayOrCollectionOfPrimitives(entryValue)) {
+          entryValue = JSON.toJSONString(entryValue);
+        }
+        properties.add(new LPGPropertyRecord(entry.getKey(), entryValue));
       }
       VertexRecord vertexRecord = new VertexRecord(node.getId(), label, properties);
       vertexRecords.add(vertexRecord);
@@ -226,7 +225,11 @@ private void writeEdge(SubGraphRecord.Edge edge) {
       List<EdgeRecord> edgeRecords = Lists.newArrayList();
       List<LPGPropertyRecord> properties = Lists.newArrayList();
       for (Map.Entry<String, Object> entry : edge.getProperties().entrySet()) {
-        properties.add(new LPGPropertyRecord(entry.getKey(), entry.getValue()));
+        Object entryValue = entry.getValue();
+        if (!TypeChecker.isArrayOrCollectionOfPrimitives(entryValue)) {
+          entryValue = JSON.toJSONString(entryValue);
+        }
+        properties.add(new LPGPropertyRecord(entry.getKey(), entryValue));
       }
 
       EdgeTypeName edgeTypeName =

cloudext/impl/graph-store/tugraph/src/main/java/com/antgroup/openspg/cloudext/impl/graphstore/tugraph/TuGraphStoreClient.java

@@ -48,7 +48,6 @@
 import com.antgroup.openspg.cloudext.interfaces.graphstore.model.lpg.schema.operation.SchemaAtomicOperationEnum;
 import com.antgroup.openspg.cloudext.interfaces.graphstore.util.TypeNameUtils;
 import com.antgroup.openspg.core.schema.model.type.BasicTypeEnum;
-import com.antgroup.openspg.server.api.facade.ApiConstants;
 import com.antgroup.tugraph.TuGraphDbRpcClient;
 import com.google.common.collect.Lists;
 import java.io.InputStream;
@@ -78,13 +77,16 @@ public class TuGraphStoreClient extends BaseLPGGraphStoreClient {
   @Getter private final LPGTypeNameConvertor typeNameConvertor;
   @Getter private final String connUrl;
 
+  private static final String ACCESS_ID = "accessId";
+  private static final String ACCESS_KEY = "accessKey";
+  private static final String TIMEOUT = "timeout";
+
   public TuGraphStoreClient(String connUrl, LPGTypeNameConvertor typeNameConvertor) {
     UriComponents uriComponents = UriComponentsBuilder.fromUriString(connUrl).build();
     this.connUrl = connUrl;
     this.graphName = uriComponents.getQueryParams().getFirst(TuGraphConstants.GRAPH_NAME);
     this.timeout =
-        Double.parseDouble(
-            String.valueOf(uriComponents.getQueryParams().getFirst(ApiConstants.TIMEOUT)));
+        Double.parseDouble(String.valueOf(uriComponents.getQueryParams().getFirst(TIMEOUT)));
     this.client = initTuGraphClient(uriComponents);
     this.internalIdGenerator = new NoChangedIdGenerator();
     this.typeNameConvertor = typeNameConvertor;
@@ -149,8 +151,8 @@ public void close() throws Exception {
 
   private TuGraphDbRpcClient initTuGraphClient(UriComponents uriComponents) {
     String host = String.format("%s:%s", uriComponents.getHost(), uriComponents.getPort());
-    String accessId = uriComponents.getQueryParams().getFirst(ApiConstants.ACCESS_ID);
-    String accessKey = uriComponents.getQueryParams().getFirst(ApiConstants.ACCESS_KEY);
+    String accessId = uriComponents.getQueryParams().getFirst(ACCESS_ID);
+    String accessKey = uriComponents.getQueryParams().getFirst(ACCESS_KEY);
     TuGraphDbRpcClient client = null;
     try {
       client = new TuGraphDbRpcClient(host, accessId, accessKey);

cloudext/impl/object-storage/minio/src/main/java/com/antgroup/openspg/cloudext/impl/objectstorage/minio/MinioClient.java

@@ -37,7 +37,9 @@
 import java.io.OutputStream;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.List;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.compress.utils.IOUtils;
@@ -72,6 +74,7 @@ private io.minio.MinioClient initMinioClient(UriComponents uriComponents) {
 
   @Override
   public Boolean saveData(String bucketName, byte[] data, String fileKey) {
+    Long start = System.currentTimeMillis();
     ByteArrayInputStream inputStream = null;
     try {
       inputStream = new ByteArrayInputStream(data);
@@ -87,19 +90,23 @@ public Boolean saveData(String bucketName, byte[] data, String fileKey) {
       log.error(e.getMessage(), e);
       throw new RuntimeException("minio saveData Exception:" + e.getMessage(), e);
     } finally {
+      log.info("minio saveData cons:{} fileKey:{}", System.currentTimeMillis() - start, fileKey);
       IOUtils.closeQuietly(inputStream);
     }
   }
 
   @Override
   public byte[] getData(String bucketName, String fileKey) {
+    Long start = System.currentTimeMillis();
     try {
       GetObjectArgs getObjectArgs =
           GetObjectArgs.builder().bucket(bucketName).object(fileKey).build();
       return inputStreamToByteArray(minioClient.getObject(getObjectArgs));
     } catch (Exception e) {
       log.error(e.getMessage(), e);
       throw new RuntimeException("minio getData Exception:" + e.getMessage(), e);
+    } finally {
+      log.info("minio getData cons:{} fileKey:{}", System.currentTimeMillis() - start, fileKey);
     }
   }
 
@@ -239,10 +246,14 @@ public String getUrlWithoutExpiration(String bucketName, String fileKey) {
   @Override
   public Boolean removeObject(String bucketName, String fileKey) {
     try {
-      RemoveObjectArgs removeObjectArgs =
-          RemoveObjectArgs.builder().bucket(bucketName).object(fileKey).build();
-      minioClient.removeObject(removeObjectArgs);
-      return true;
+      if (isDirectory(bucketName, fileKey)) {
+        return removeDirectory(bucketName, fileKey);
+      } else {
+        RemoveObjectArgs removeObjectArgs =
+            RemoveObjectArgs.builder().bucket(bucketName).object(fileKey).build();
+        minioClient.removeObject(removeObjectArgs);
+        return true;
+      }
     } catch (Exception e) {
       log.error(e.getMessage(), e);
       throw new RuntimeException("minio removeObject Exception:" + e.getMessage(), e);
@@ -252,26 +263,64 @@ public Boolean removeObject(String bucketName, String fileKey) {
   @Override
   public Boolean removeDirectory(String bucketName, String directoryPath) {
     try {
-      Iterable<Result<Item>> objects =
+      List<String> files = getAllFilesRecursively(bucketName, directoryPath);
+
+      for (String file : files) {
+        log.info("minio Deleting: " + file);
+        if (file.startsWith(directoryPath)) {
+          removeObject(bucketName, file);
+        }
+      }
+      return true;
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+      throw new RuntimeException("minio removeDirectory Exception:" + e.getMessage(), e);
+    }
+  }
+
+  @Override
+  public Boolean isDirectory(String bucketName, String path) {
+    try {
+      Iterable<Result<Item>> items =
+          minioClient.listObjects(
+              ListObjectsArgs.builder().bucket(bucketName).prefix(path).recursive(false).build());
+      for (Result<Item> result : items) {
+        Item item = result.get();
+        if (item.isDir() || !item.objectName().equals(path)) {
+          return true;
+        }
+      }
+      return false;
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+      throw new RuntimeException("minio isDirectory Exception:" + e.getMessage(), e);
+    }
+  }
+
+  @Override
+  public List<String> getAllFilesRecursively(String bucketName, String directoryPath) {
+    List<String> filePaths = new ArrayList<>();
+    try {
+      Iterable<Result<Item>> items =
           minioClient.listObjects(
               ListObjectsArgs.builder()
                   .bucket(bucketName)
                   .prefix(directoryPath)
-                  .recursive(true)
+                  .recursive(false)
                   .build());
-
-      for (Result<Item> result : objects) {
+      for (Result<Item> result : items) {
         Item item = result.get();
-        log.info("minio Deleting: " + item.objectName());
-        if (item.objectName().startsWith(directoryPath)) {
-          removeObject(bucketName, item.objectName());
+        if (item.isDir()) {
+          filePaths.addAll(getAllFilesRecursively(bucketName, item.objectName()));
+        } else {
+          filePaths.add(item.objectName());
         }
       }
-      return true;
     } catch (Exception e) {
       log.error(e.getMessage(), e);
-      throw new RuntimeException("minio removeDirectory Exception:" + e.getMessage(), e);
+      throw new RuntimeException("minio getAllFilesRecursively Exception: " + e.getMessage(), e);
     }
+    return filePaths;
   }
 
   @Override
@@ -287,6 +336,27 @@ public Long getContentLength(String bucketName, String objectName) {
     }
   }
 
+  @Override
+  public long getStorageSize(String bucketName, String path) {
+    try {
+      Iterable<Result<Item>> items =
+          minioClient.listObjects(
+              ListObjectsArgs.builder().bucket(bucketName).prefix(path).recursive(true).build());
+
+      long totalSizeInBytes = 0;
+      for (Result<Item> itemResult : items) {
+        Item item = itemResult.get();
+        if (!item.isDir()) {
+          totalSizeInBytes += item.size();
+        }
+      }
+      return totalSizeInBytes;
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+      throw new RuntimeException("minio getStorageSize Exception:" + e.getMessage(), e);
+    }
+  }
+
   public void makeBucket(String bucketName) {
     try {
       boolean isExist =

cloudext/impl/object-storage/oss/src/main/java/com/antgroup/openspg/cloudext/impl/objectstorage/oss/OSSClient.java

@@ -33,7 +33,9 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.List;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.util.UriComponents;
@@ -42,6 +44,8 @@
 @Slf4j
 public class OSSClient implements ObjectStorageClient {
 
+  private static final String SLASH = "/";
+
   private final OSS ossClient;
 
   @Getter private final String connUrl;
@@ -209,8 +213,12 @@ public String getUrlWithoutExpiration(String bucketName, String fileKey) {
   @Override
   public Boolean removeObject(String bucketName, String fileKey) {
     try {
-      ossClient.deleteObject(bucketName, fileKey);
-      return true;
+      if (isDirectory(bucketName, fileKey)) {
+        return removeDirectory(bucketName, fileKey);
+      } else {
+        ossClient.deleteObject(bucketName, fileKey);
+        return true;
+      }
     } catch (Exception e) {
       log.error(e.getMessage(), e);
       throw new RuntimeException("OSS removeObject Exception:" + e.getMessage(), e);
@@ -220,25 +228,67 @@ public Boolean removeObject(String bucketName, String fileKey) {
   @Override
   public Boolean removeDirectory(String bucketName, String directoryPath) {
     try {
-      ObjectListing objectListing;
+      List<String> files = getAllFilesRecursively(bucketName, directoryPath);
+
+      for (String file : files) {
+        log.info("OSS Deleting: " + file);
+        if (file.startsWith(directoryPath)) {
+          removeObject(bucketName, file);
+        }
+      }
+      return true;
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+      throw new RuntimeException("OSS removeDirectory Exception: " + e.getMessage(), e);
+    }
+  }
 
+  @Override
+  public Boolean isDirectory(String bucketName, String path) {
+    try {
+      ObjectListing objectListing = ossClient.listObjects(bucketName, path);
+      if (!objectListing.getCommonPrefixes().isEmpty()
+          || !objectListing.getObjectSummaries().isEmpty()) {
+        if (path.endsWith(SLASH)) {
+          return true;
+        }
+      }
+      for (OSSObjectSummary objectSummary : objectListing.getObjectSummaries()) {
+        String objectName = objectSummary.getKey();
+        if (objectName.endsWith(SLASH) || !objectName.equals(path)) {
+          return true;
+        }
+      }
+      return false;
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+      throw new RuntimeException("OSS isDirectory Exception: " + e.getMessage(), e);
+    }
+  }
+
+  @Override
+  public List<String> getAllFilesRecursively(String bucketName, String directoryPath) {
+    List<String> filePaths = new ArrayList<>();
+    try {
+      String nextMarker;
       do {
-        objectListing = ossClient.listObjects(bucketName, directoryPath);
-        for (OSSObjectSummary objectSummary : objectListing.getObjectSummaries()) {
-          String objectName = objectSummary.getKey();
-          log.info("OSS Deleting: " + objectName);
-          if (objectName.startsWith(directoryPath)) {
-            removeObject(bucketName, objectName);
+        ObjectListing objectListing = ossClient.listObjects(bucketName, directoryPath);
+        List<OSSObjectSummary> objectSummaries = objectListing.getObjectSummaries();
+        for (OSSObjectSummary summary : objectSummaries) {
+          if (summary.getKey().endsWith(SLASH)) {
+            filePaths.addAll(getAllFilesRecursively(bucketName, summary.getKey()));
+          } else {
+            filePaths.add(summary.getKey());
           }
         }
-        objectListing.getNextMarker();
-      } while (objectListing.isTruncated());
+        nextMarker = objectListing.getNextMarker();
+      } while (nextMarker != null);
 
-      return true;
     } catch (Exception e) {
-      log.error(e.getMessage(), e);
-      throw new RuntimeException("OSS removeDirectory Exception:" + e.getMessage(), e);
+      log.error("OSS getAllFilesRecursively Exception: " + e.getMessage(), e);
+      throw new RuntimeException("OSS getAllFilesRecursively Exception: " + e.getMessage(), e);
     }
+    return filePaths;
   }
 
   @Override
@@ -251,4 +301,25 @@ public Long getContentLength(String bucketName, String objectName) {
       throw new RuntimeException("OSS getContentLength Exception:" + e.getMessage(), e);
     }
   }
+
+  @Override
+  public long getStorageSize(String bucketName, String path) {
+    long totalSizeInBytes = 0L;
+
+    try {
+      String nextMarker;
+      do {
+        ObjectListing objectListing = ossClient.listObjects(bucketName, path);
+        for (OSSObjectSummary objectSummary : objectListing.getObjectSummaries()) {
+          totalSizeInBytes += objectSummary.getSize();
+        }
+        nextMarker = objectListing.getNextMarker();
+      } while (nextMarker != null);
+      return totalSizeInBytes;
+
+    } catch (Exception e) {
+      log.error(e.getMessage(), e);
+      throw new RuntimeException("OSS getStorageSize Exception:" + e.getMessage(), e);
+    }
+  }
 }