You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by ji...@apache.org on 2026/08/12 04:46:03 UTC
(hugegraph-toolchain) branch master updated: refactor(hubble): correct query console, list rendering and load lifecycle defects (#755)
This is an automated email from the ASF dual-hosted git repository.
imbajin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hugegraph-toolchain.git
The following commit(s) were added to refs/heads/master by this push:
new 83acb4d7b refactor(hubble): correct query console, list rendering and load lifecycle defects (#755)
83acb4d7b is described below
commit 83acb4d7bef9e5293eb18fe5abe50a75ccfbb685
Author: KAI <hi...@gmail.com>
AuthorDate: Wed Aug 12 10:15:58 2026 +0530
refactor(hubble): correct query console, list rendering and load lifecycle defects (#755)
Bug fixes for Hubble found during a focused exploration pass, followed by a code audit of the surrounding paths. 10 commits, 62 files.
Findings were originally gathered against a single deployment shape: the all-in-one hugegraph/hugegraph:latest image, RocksDB backend, standalone mode with PD disabled. The fixes have since been re-validated against a distributed deployment (3 PD, 3 Store, 3 Server, PR-head Hubble).
The standalone fixes are complete and the repository-native regression baseline remains green. The live distributed round found no distributed-specific production compatibility correction was needed; its first follow-up added a regression test for the negative page_size rejection. Two later bot-review commits narrowly harden concurrency, migration, stale-response, retention, and launcher process-identity edges in those same documented fixed paths.
---------
Co-authored-by: dark <ji...@apache.org>
---
.../org/apache/hugegraph/driver/CypherManager.java | 12 +-
.../org/apache/hugegraph/driver/HugeClient.java | 33 +-
.../hugegraph/config/DatabaseSchemaMigrator.java | 609 ++++++++++++++++++++-
.../controller/graph/GraphController.java | 54 +-
.../controller/ingest/IngestController.java | 22 +
.../controller/load/FileMappingController.java | 40 +-
.../controller/load/FileUploadController.java | 72 ++-
.../controller/load/JobManagerController.java | 10 +-
.../controller/load/LoadTaskController.java | 20 +-
.../hugegraph/controller/op/HStoreController.java | 48 +-
.../hugegraph/controller/op/LogController.java | 10 +-
.../org/apache/hugegraph/entity/load/LoadTask.java | 14 +-
.../hugegraph/entity/query/ApplicationInfo.java | 4 +
.../hugegraph/entity/query/ElementEditHistory.java | 2 +
.../apache/hugegraph/handler/ExceptionAdvisor.java | 7 +-
.../hugegraph/mapper/load/FileMappingMapper.java | 7 +
.../hugegraph/mapper/load/JobManagerMapper.java | 14 +-
.../mapper/query/EditElementHistoryMapper.java | 8 +-
.../mapper/query/ExecuteHistoryMapper.java | 53 +-
.../hugegraph/service/graphs/GraphsService.java | 61 ++-
.../hugegraph/service/load/DatasourceService.java | 4 +-
.../hugegraph/service/load/FileMappingService.java | 87 ++-
.../hugegraph/service/load/JobManagerService.java | 45 +-
.../hugegraph/service/load/LoadTaskService.java | 74 ++-
.../org/apache/hugegraph/service/op/ESService.java | 7 +-
.../apache/hugegraph/service/op/LogService.java | 6 +-
.../service/query/EditElementHistoryService.java | 5 +-
.../service/query/ExecuteHistoryService.java | 7 +-
.../service/query/GremlinCollectionService.java | 2 +-
.../hugegraph/service/query/QueryService.java | 107 +++-
.../org/apache/hugegraph/util/GremlinUtil.java | 190 ++++++-
.../java/org/apache/hugegraph/util/PageUtil.java | 24 +
.../src/main/resources/database/schema.sql | 7 +-
.../src/main/resources/i18n/messages.properties | 25 +-
.../main/resources/i18n/messages_zh_CN.properties | 23 +
.../hugegraph/config/LegacySchemaStartupTest.java | 233 ++++++++
.../controller/op/HStoreControllerTest.java | 114 ++++
.../load/IngestTransactionIntegrationTest.java | 150 +++++
.../service/query/ExecuteHistoryRetentionTest.java | 188 +++++++
.../hugegraph/unit/FileMappingDeletionTest.java | 110 ++++
.../hugegraph/unit/FileMappingSchemaTest.java | 276 ++++++++++
.../hugegraph/unit/FileUploadControllerTest.java | 147 +++++
.../org/apache/hugegraph/unit/GremlinUtilTest.java | 190 +++++++
.../hugegraph/unit/LoaderScopeControllerTest.java | 11 +-
.../org/apache/hugegraph/unit/PriorityFixTest.java | 20 +
.../org/apache/hugegraph/unit/UnitTestSuite.java | 1 +
.../assembly/static/bin/start-hubble.sh | 46 +-
.../hubble-dist/assembly/static/bin/stop-hubble.sh | 70 ++-
hugegraph-hubble/hubble-fe/src/api/request.js | 17 +
.../src/i18n/resources/en-US/modules/analysis.json | 1 +
.../src/i18n/resources/en-US/modules/pages.json | 2 +
.../src/i18n/resources/zh-CN/modules/analysis.json | 1 +
.../src/i18n/resources/zh-CN/modules/pages.json | 2 +
.../hubble-fe/src/modules/algorithm/Home/index.js | 14 +
.../hubble-fe/src/modules/analysis/Home/index.js | 12 +-
.../src/modules/asyncTasks/Detail/index.js | 2 +-
.../src/modules/component/DynamicAddEdge/index.js | 2 +-
.../src/modules/component/ImportData/index.js | 25 +-
.../hubble-fe/src/pages/Datasource/EditLayer.js | 55 +-
.../src/pages/Datasource/EditLayer.test.js | 83 ++-
.../hubble-fe/src/pages/Graph/index.js | 5 +-
.../hubble-fe/src/pages/Graph/index.module.scss | 4 +
.../hubble-fe/src/pages/GraphSpace/Card.js | 2 +-
.../hubble-fe/src/pages/Meta/Edge/index.js | 5 +-
.../hubble-fe/src/pages/Meta/Vertex/index.js | 3 +-
hugegraph-hubble/hubble-fe/src/pages/My/index.js | 6 +-
.../hubble-fe/src/pages/Operations/Nodes.js | 4 +-
.../src/pages/Task/components/EditLayer.js | 44 +-
hugegraph-hubble/hubble-fe/src/pages/Task/index.js | 13 +-
.../hubble-fe/src/pages/TaskDetail/index.js | 7 +-
.../hubble-fe/src/pages/TaskEdit/BaseForm/index.js | 13 +-
.../hubble-fe/src/utils/formatTimeDuration.js | 13 +-
72 files changed, 3361 insertions(+), 243 deletions(-)
diff --git a/hugegraph-client/src/main/java/org/apache/hugegraph/driver/CypherManager.java b/hugegraph-client/src/main/java/org/apache/hugegraph/driver/CypherManager.java
index 498e568a0..2cf9e483e 100644
--- a/hugegraph-client/src/main/java/org/apache/hugegraph/driver/CypherManager.java
+++ b/hugegraph-client/src/main/java/org/apache/hugegraph/driver/CypherManager.java
@@ -18,6 +18,7 @@
package org.apache.hugegraph.driver;
import org.apache.hugegraph.api.gremlin.CypherAPI;
+import org.apache.hugegraph.exception.ServerException;
import org.apache.hugegraph.api.job.CypherJobAPI;
import org.apache.hugegraph.client.RestClient;
import org.apache.hugegraph.structure.gremlin.Response;
@@ -47,8 +48,17 @@ public class CypherManager {
public ResultSet execute(String cypher) {
Response response = this.cypherAPI.post(this.graphSpace, this.graph, cypher);
+ Response.Status status = response.status();
+ if (status != null && status.code() != 200) {
+ String message = status.message();
+ if (message == null || message.isEmpty()) {
+ message = "Cypher query failed with status code " + status.code();
+ }
+ ServerException exception = new ServerException(message);
+ exception.status(status.code());
+ throw exception;
+ }
response.graphManager(this.graphManager);
- // TODO: Can add some checks later
return response.result();
}
diff --git a/hugegraph-client/src/main/java/org/apache/hugegraph/driver/HugeClient.java b/hugegraph-client/src/main/java/org/apache/hugegraph/driver/HugeClient.java
index 6327b0023..3f26091f3 100644
--- a/hugegraph-client/src/main/java/org/apache/hugegraph/driver/HugeClient.java
+++ b/hugegraph-client/src/main/java/org/apache/hugegraph/driver/HugeClient.java
@@ -50,6 +50,13 @@ public class HugeClient implements Closeable {
private final boolean borrowedClient;
private final RestClient client;
+ /*
+ * Whether the server api-version handshake has been performed for
+ * this client instance, so that it runs only once instead of on
+ * every assignGraph() call
+ */
+ private volatile boolean apiVersionChecked;
+ private final Object apiVersionLock = new Object();
private VersionManager version;
private GraphsManager graphs;
private SchemaManager schema;
@@ -145,9 +152,18 @@ public class HugeClient implements Closeable {
public void initManagers(RestClient client, String graphSpace,
String graph) {
assert client != null;
- // Check hugegraph-server api version
+ // Check hugegraph-server api version (once per client instance)
this.version = new VersionManager(client);
- this.checkServerApiVersion();
+ if (!this.apiVersionChecked) {
+ synchronized (this.apiVersionLock) {
+ if (!this.apiVersionChecked) {
+ this.checkServerApiVersion();
+ // Failed handshakes leave the flag false so a later
+ // assignment can retry.
+ this.apiVersionChecked = true;
+ }
+ }
+ }
this.graphs = new GraphsManager(client, graphSpace);
this.auth = new AuthManager(client, graphSpace, graph);
@@ -171,6 +187,19 @@ public class HugeClient implements Closeable {
this.variable = new VariablesManager(client, graphSpace, graph);
this.job = new JobManager(client, graphSpace, graph);
this.task = new TaskManager(client, graphSpace, graph);
+ } else {
+ /*
+ * No graph is assigned: reset all graph-scoped managers so
+ * that stale managers from a previous graph cannot be used
+ */
+ this.schema = null;
+ this.graph = null;
+ this.gremlin = null;
+ this.cypher = null;
+ this.traverser = null;
+ this.variable = null;
+ this.job = null;
+ this.task = null;
}
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/config/DatabaseSchemaMigrator.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/config/DatabaseSchemaMigrator.java
index a95bad3fd..235c36bb7 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/config/DatabaseSchemaMigrator.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/config/DatabaseSchemaMigrator.java
@@ -24,13 +24,21 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Locale;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
import javax.sql.DataSource;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanInitializationException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.core.Ordered;
+import org.springframework.core.PriorityOrdered;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -42,27 +50,75 @@ import lombok.extern.log4j.Log4j2;
@Log4j2
@Component
-public class DatabaseSchemaMigrator implements ApplicationRunner {
+public class DatabaseSchemaMigrator implements BeanPostProcessor,
+ PriorityOrdered {
private static final int FILE_MAPPING_PATH_LENGTH = 2048;
+ private static final int COLLECTION_NAME_LENGTH = 48;
+ private static final String DEFAULT_GRAPHSPACE = "DEFAULT";
+ private static final String DEFAULT_GRAPH = "hugegraph";
+ private static final String GREMLIN_TYPE = "GREMLIN";
private static final String FILE_MAPPING_TABLE = "file_mapping";
private static final String FILE_MAPPING_PATH_COLUMN = "path";
private static final String EXECUTE_HISTORY_TABLE = "execute_history";
private static final String FAILURE_REASON_COLUMN = "failure_reason";
private static final String LOAD_TASK_TABLE = "load_task";
+ private static final String GREMLIN_COLLECTION_TABLE =
+ "gremlin_collection";
+ private static final String GREMLIN_COLLECTION_SCOPE_INDEX =
+ "gremlin_collection_scope_name";
+ /*
+ * Columns added to `execute_history` after the legacy schema was
+ * released. Each entry is {column, type, backfill value or null}.
+ */
+ private static final String[][] EXECUTE_HISTORY_COLUMNS = {
+ {"conn_id", "INT", null},
+ {"graphspace", "VARCHAR(48)", null},
+ {"graph", "VARCHAR(48)", null},
+ {"async_id", "BIGINT", "0"},
+ {"text", "TEXT", "''"},
+ {"async_status", "TINYINT", "0"}
+ };
+ /* Columns added to `gremlin_collection` after the legacy schema. */
+ private static final String[][] GREMLIN_COLLECTION_COLUMNS = {
+ {"conn_id", "INT", null},
+ {"graphspace", "VARCHAR(48)", null},
+ {"graph", "VARCHAR(48)", null},
+ {"type", "VARCHAR(48)", null}
+ };
private static final String[] LOAD_OPTION_CREDENTIALS = {
"password", "token", "pdToken", "trustStoreToken"
};
+ private static final String[][] REQUIRED_INDEXES = {
+ {"execute_history", "execute_history_graph_create_time",
+ "graphspace", "graph", "create_time"},
+ {"file_mapping", "file_mapping_job_id", "job_id"},
+ {"load_task", "load_task_job_id", "job_id"},
+ {"job_manager", "job_manager_graph_create_time",
+ "graphspace", "graph", "create_time"},
+ {"async_task", "async_task_graph", "graphspace", "graph"}
+ };
private static final ObjectMapper JSON = new ObjectMapper();
- @Autowired
- private DataSource dataSource;
+ @Override
+ public int getOrder() {
+ // Boot's DataSourceInitializerPostProcessor runs at HIGHEST + 1.
+ return Ordered.HIGHEST_PRECEDENCE;
+ }
@Override
- public void run(ApplicationArguments args) throws Exception {
- try (Connection conn = this.dataSource.getConnection()) {
+ public Object postProcessAfterInitialization(Object bean, String beanName)
+ throws BeansException {
+ if (!(bean instanceof DataSource)) {
+ return bean;
+ }
+ try (Connection conn = ((DataSource) bean).getConnection()) {
this.migrate(conn);
+ } catch (SQLException e) {
+ throw new BeanInitializationException(
+ "Failed to migrate Hubble database schema", e);
}
+ return bean;
}
public void migrate(Connection conn) throws SQLException {
@@ -73,9 +129,487 @@ public class DatabaseSchemaMigrator implements ApplicationRunner {
}
this.migrateExecuteHistoryFailureReason(conn);
+ this.addMissingColumns(conn, EXECUTE_HISTORY_TABLE,
+ EXECUTE_HISTORY_COLUMNS);
+ this.addMissingColumns(conn, GREMLIN_COLLECTION_TABLE,
+ GREMLIN_COLLECTION_COLUMNS);
+ this.migrateLegacyQueryScopes(conn);
+ this.migrateGremlinCollectionUniqueConstraint(conn);
+ this.addMissingIndexes(conn);
this.removeLegacyLoadTaskCredentials(conn);
}
+ private void migrateLegacyQueryScopes(Connection conn)
+ throws SQLException {
+ List<LegacyScope> scopes = this.legacyScopes(conn);
+ LegacyScope fallback = scopes.isEmpty() ?
+ new LegacyScope(null, DEFAULT_GRAPHSPACE,
+ DEFAULT_GRAPH) : scopes.get(0);
+ int updated = 0;
+ if (this.scopeColumnsExist(conn, EXECUTE_HISTORY_TABLE)) {
+ updated += this.backfillScopes(conn, EXECUTE_HISTORY_TABLE, scopes,
+ fallback);
+ }
+ if (this.scopeColumnsExist(conn, GREMLIN_COLLECTION_TABLE) &&
+ this.columnExists(conn, GREMLIN_COLLECTION_TABLE, "type")) {
+ updated += this.backfillScopes(conn, GREMLIN_COLLECTION_TABLE,
+ scopes, fallback);
+ try (PreparedStatement statement = conn.prepareStatement(
+ "UPDATE `gremlin_collection` SET `type` = ? " +
+ "WHERE `type` IS NULL OR `type` = ''")) {
+ statement.setString(1, GREMLIN_TYPE);
+ updated += statement.executeUpdate();
+ }
+ }
+ if (updated > 0) {
+ log.info("Recovered scope for {} legacy query records", updated);
+ }
+ }
+
+ boolean scopeColumnsExist(Connection conn, String table)
+ throws SQLException {
+ return this.tableExists(conn, table) &&
+ this.columnExists(conn, table, "conn_id") &&
+ this.columnExists(conn, table, "graphspace") &&
+ this.columnExists(conn, table, "graph");
+ }
+
+ private int backfillScopes(Connection conn, String table,
+ List<LegacyScope> scopes,
+ LegacyScope fallback) throws SQLException {
+ int updated = 0;
+ for (LegacyScope scope : scopes) {
+ updated += this.backfillConnectionScope(conn, table, scope);
+ }
+ try (PreparedStatement statement = conn.prepareStatement(
+ String.format("UPDATE `%s` SET `graphspace` = ? " +
+ "WHERE `graphspace` IS NULL OR " +
+ "`graphspace` = ''", table))) {
+ statement.setString(1, fallback.graphspace);
+ updated += statement.executeUpdate();
+ }
+ try (PreparedStatement statement = conn.prepareStatement(
+ String.format("UPDATE `%s` SET `graph` = ? " +
+ "WHERE `graph` IS NULL OR `graph` = ''",
+ table))) {
+ statement.setString(1, fallback.graph);
+ updated += statement.executeUpdate();
+ }
+ return updated;
+ }
+
+ private int backfillConnectionScope(Connection conn, String table,
+ LegacyScope scope) throws SQLException {
+ int updated = 0;
+ try (PreparedStatement statement = conn.prepareStatement(
+ String.format("UPDATE `%s` SET `graphspace` = ? WHERE " +
+ "`conn_id` = ? AND (`graphspace` IS NULL OR " +
+ "`graphspace` = '')", table))) {
+ statement.setString(1, scope.graphspace);
+ statement.setInt(2, scope.connectionId);
+ updated += statement.executeUpdate();
+ }
+ try (PreparedStatement statement = conn.prepareStatement(
+ String.format("UPDATE `%s` SET `graph` = ? WHERE " +
+ "`conn_id` = ? AND (`graph` IS NULL OR " +
+ "`graph` = '')", table))) {
+ statement.setString(1, scope.graph);
+ statement.setInt(2, scope.connectionId);
+ updated += statement.executeUpdate();
+ }
+ return updated;
+ }
+
+ private List<LegacyScope> legacyScopes(Connection conn)
+ throws SQLException {
+ List<LegacyScope> scopes = new ArrayList<>();
+ if (!this.tableExists(conn, "graph_connection") ||
+ !this.columnExists(conn, "graph_connection", "id") ||
+ !this.columnExists(conn, "graph_connection", "graph")) {
+ return scopes;
+ }
+
+ boolean hasGraphspace = this.columnExists(conn, "graph_connection",
+ "graphspace");
+ String sql = hasGraphspace ?
+ "SELECT `id`, `graphspace`, `graph` FROM " +
+ "`graph_connection` ORDER BY `id`" :
+ "SELECT `id`, `graph` FROM `graph_connection` " +
+ "ORDER BY `id`";
+ try (Statement statement = conn.createStatement();
+ ResultSet rows = statement.executeQuery(sql)) {
+ while (rows.next()) {
+ Integer connectionId = rows.getInt(1);
+ String graphspace = hasGraphspace ? rows.getString(2) : null;
+ String graph = rows.getString(hasGraphspace ? 3 : 2);
+ scopes.add(new LegacyScope(
+ connectionId,
+ blank(graphspace) ? DEFAULT_GRAPHSPACE : graphspace,
+ blank(graph) ? DEFAULT_GRAPH : graph));
+ }
+ }
+ return scopes;
+ }
+
+ private void migrateGremlinCollectionUniqueConstraint(Connection conn)
+ throws SQLException {
+ if (!this.tableExists(conn, GREMLIN_COLLECTION_TABLE) ||
+ !this.columnExists(conn, GREMLIN_COLLECTION_TABLE,
+ "graphspace") ||
+ !this.columnExists(conn, GREMLIN_COLLECTION_TABLE, "graph") ||
+ !this.columnExists(conn, GREMLIN_COLLECTION_TABLE, "name") ||
+ !this.columnExists(conn, GREMLIN_COLLECTION_TABLE, "type")) {
+ return;
+ }
+
+ String product = conn.getMetaData().getDatabaseProductName();
+ String normalizedProduct = product.toLowerCase(Locale.ROOT);
+ if (!normalizedProduct.contains("h2") &&
+ !normalizedProduct.contains("mysql") &&
+ !normalizedProduct.contains("mariadb")) {
+ log.warn("Skip gremlin_collection unique-key migration for " +
+ "unsupported database product {}", product);
+ return;
+ }
+ this.renameLegacyCollectionCollisions(conn);
+ List<String> obsolete = normalizedProduct.contains("h2") ?
+ this.h2ObsoleteUniqueConstraints(conn) :
+ this.mysqlObsoleteUniqueIndexes(conn, product);
+ for (String name : obsolete) {
+ try (Statement statement = conn.createStatement()) {
+ statement.execute(dropUniqueKeySql(product, name));
+ }
+ log.info("Removed obsolete gremlin_collection unique key {}",
+ name);
+ }
+
+ Map<String, List<String>> uniqueIndexes = this.uniqueIndexes(conn,
+ GREMLIN_COLLECTION_TABLE);
+ if (this.hasScopeUniqueIndex(uniqueIndexes)) {
+ return;
+ }
+ try (Statement statement = conn.createStatement()) {
+ statement.execute("CREATE UNIQUE INDEX `" +
+ GREMLIN_COLLECTION_SCOPE_INDEX + "` ON `" +
+ GREMLIN_COLLECTION_TABLE + "`(" +
+ "`graphspace`, `graph`, `name`, `type`)");
+ }
+ log.info("Added scoped gremlin_collection unique index {}",
+ GREMLIN_COLLECTION_SCOPE_INDEX);
+ }
+
+ private void renameLegacyCollectionCollisions(Connection conn)
+ throws SQLException {
+ List<LegacyCollectionCollision> collisions = new ArrayList<>();
+ try (Statement statement = conn.createStatement();
+ ResultSet rows = statement.executeQuery(
+ "SELECT `graphspace`, `graph`, `name`, `type` FROM " +
+ "`gremlin_collection` GROUP BY `graphspace`, `graph`, " +
+ "`name`, `type` HAVING COUNT(*) > 1")) {
+ while (rows.next()) {
+ collisions.add(new LegacyCollectionCollision(
+ rows.getString(1), rows.getString(2),
+ rows.getString(3), rows.getString(4)));
+ }
+ }
+
+ for (LegacyCollectionCollision collision : collisions) {
+ List<LegacyCollectionRow> duplicateRows =
+ this.legacyCollectionRows(conn, collision);
+ for (int i = 1; i < duplicateRows.size(); i++) {
+ LegacyCollectionRow row = duplicateRows.get(i);
+ String name = this.uniqueLegacyCollectionName(conn, collision,
+ row);
+ try (PreparedStatement statement = conn.prepareStatement(
+ "UPDATE `gremlin_collection` SET `name` = ? " +
+ "WHERE `id` = ?")) {
+ statement.setString(1, name);
+ statement.setInt(2, row.id);
+ statement.executeUpdate();
+ }
+ log.warn("Renamed colliding legacy saved query {} (id {}) " +
+ "to {} while recovering graph scope", collision.name,
+ row.id, name);
+ }
+ }
+ }
+
+ private List<LegacyCollectionRow> legacyCollectionRows(
+ Connection conn, LegacyCollectionCollision collision)
+ throws SQLException {
+ List<LegacyCollectionRow> rows = new ArrayList<>();
+ try (PreparedStatement statement = conn.prepareStatement(
+ "SELECT `id`, `conn_id` FROM `gremlin_collection` WHERE " +
+ "`graphspace` = ? AND `graph` = ? AND `name` = ? AND " +
+ "`type` = ? ORDER BY `id`")) {
+ statement.setString(1, collision.graphspace);
+ statement.setString(2, collision.graph);
+ statement.setString(3, collision.name);
+ statement.setString(4, collision.type);
+ try (ResultSet result = statement.executeQuery()) {
+ while (result.next()) {
+ int connectionId = result.getInt(2);
+ rows.add(new LegacyCollectionRow(
+ result.getInt(1),
+ result.wasNull() ? null : connectionId));
+ }
+ }
+ }
+ return rows;
+ }
+
+ private String uniqueLegacyCollectionName(
+ Connection conn, LegacyCollectionCollision collision,
+ LegacyCollectionRow row) throws SQLException {
+ for (int attempt = 0; ; attempt++) {
+ String suffix = "_" + row.id +
+ (attempt == 0 ? "" : "_" + attempt);
+ int baseLength = Math.min(collision.name.length(),
+ COLLECTION_NAME_LENGTH - suffix.length());
+ String candidate = collision.name.substring(0, baseLength) + suffix;
+ if (!this.legacyCollectionNameExists(conn, collision, row,
+ candidate)) {
+ return candidate;
+ }
+ }
+ }
+
+ private boolean legacyCollectionNameExists(
+ Connection conn, LegacyCollectionCollision collision,
+ LegacyCollectionRow row, String name) throws SQLException {
+ String connectionPredicate = row.connectionId == null ?
+ "`conn_id` IS NULL" : "`conn_id` = ?";
+ try (PreparedStatement statement = conn.prepareStatement(
+ "SELECT 1 FROM `gremlin_collection` WHERE `name` = ? AND " +
+ "((`graphspace` = ? AND `graph` = ? AND `type` = ?) OR " +
+ connectionPredicate + ")")) {
+ statement.setString(1, name);
+ statement.setString(2, collision.graphspace);
+ statement.setString(3, collision.graph);
+ statement.setString(4, collision.type);
+ if (row.connectionId != null) {
+ statement.setInt(5, row.connectionId);
+ }
+ try (ResultSet result = statement.executeQuery()) {
+ return result.next();
+ }
+ }
+ }
+
+ private List<String> h2ObsoleteUniqueConstraints(Connection conn)
+ throws SQLException {
+ List<String> constraints = new ArrayList<>();
+ try (PreparedStatement statement = conn.prepareStatement(
+ "SELECT `CONSTRAINT_NAME`, `COLUMN_LIST` FROM " +
+ "`INFORMATION_SCHEMA`.`CONSTRAINTS` WHERE " +
+ "UPPER(`TABLE_NAME`) = ? AND `CONSTRAINT_TYPE` = 'UNIQUE'")) {
+ statement.setString(1,
+ GREMLIN_COLLECTION_TABLE.toUpperCase(Locale.ROOT));
+ try (ResultSet rows = statement.executeQuery()) {
+ while (rows.next()) {
+ if (obsoleteUniqueColumns(rows.getString(2))) {
+ constraints.add(rows.getString(1));
+ }
+ }
+ }
+ }
+ return constraints;
+ }
+
+ private List<String> mysqlObsoleteUniqueIndexes(Connection conn,
+ String product)
+ throws SQLException {
+ String normalized = product.toLowerCase(Locale.ROOT);
+ if (!normalized.contains("mysql") &&
+ !normalized.contains("mariadb")) {
+ log.warn("Skip gremlin_collection unique-key migration for " +
+ "unsupported database product {}", product);
+ return new ArrayList<>();
+ }
+ List<String> indexes = new ArrayList<>();
+ for (Map.Entry<String, List<String>> entry :
+ this.uniqueIndexes(conn, GREMLIN_COLLECTION_TABLE).entrySet()) {
+ if (obsoleteUniqueColumns(String.join(",", entry.getValue()))) {
+ indexes.add(entry.getKey());
+ }
+ }
+ return indexes;
+ }
+
+ private Map<String, List<String>> uniqueIndexes(Connection conn,
+ String table)
+ throws SQLException {
+ Map<String, SortedMap<Short, String>> columns = new LinkedHashMap<>();
+ DatabaseMetaData metaData = conn.getMetaData();
+ String[] tables = {table, table.toUpperCase(Locale.ROOT)};
+ for (String tableName : tables) {
+ try (ResultSet rows = metaData.getIndexInfo(conn.getCatalog(), null,
+ tableName, true,
+ false)) {
+ while (rows.next()) {
+ String index = rows.getString("INDEX_NAME");
+ String column = rows.getString("COLUMN_NAME");
+ if (index == null || column == null ||
+ "PRIMARY".equalsIgnoreCase(index) ||
+ index.toUpperCase(Locale.ROOT).startsWith("PRIMARY_KEY")) {
+ continue;
+ }
+ short position = rows.getShort("ORDINAL_POSITION");
+ columns.computeIfAbsent(index, key -> new TreeMap<>())
+ .put(position, column.toLowerCase(Locale.ROOT));
+ }
+ }
+ if (!columns.isEmpty()) {
+ break;
+ }
+ }
+ Map<String, List<String>> indexes = new LinkedHashMap<>();
+ for (Map.Entry<String, SortedMap<Short, String>> entry :
+ columns.entrySet()) {
+ indexes.put(entry.getKey(),
+ new ArrayList<>(entry.getValue().values()));
+ }
+ return indexes;
+ }
+
+ private boolean hasScopeUniqueIndex(Map<String, List<String>> indexes) {
+ for (List<String> columns : indexes.values()) {
+ if ("graphspace,graph,name,type".equals(String.join(",", columns))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean obsoleteUniqueColumns(String columns) {
+ String normalized = columns.toLowerCase(Locale.ROOT)
+ .replace("`", "")
+ .replace(" ", "");
+ return "name".equals(normalized) ||
+ "conn_id,name".equals(normalized);
+ }
+
+ static String dropUniqueKeySql(String productName, String name) {
+ String operation = productName.toLowerCase(Locale.ROOT).contains("h2") ?
+ "DROP CONSTRAINT" : "DROP INDEX";
+ return "ALTER TABLE `gremlin_collection` " + operation + " `" +
+ name.replace("`", "``") + "`";
+ }
+
+ private static boolean blank(String value) {
+ return value == null || value.trim().isEmpty();
+ }
+
+ /**
+ * Add the columns a legacy Hubble database is missing. `CREATE TABLE IF
+ * NOT EXISTS` is a no-op against a pre-existing table, so an H2 file
+ * created by an older Hubble keeps its old column set and every query
+ * touching the newer columns fails. Each column is added nullable and
+ * then backfilled, which keeps the statement portable and idempotent:
+ * on a fresh database every column already exists and nothing is run.
+ */
+ private void addMissingColumns(Connection conn, String table,
+ String[][] columns) throws SQLException {
+ if (!this.tableExists(conn, table)) {
+ return;
+ }
+
+ String product = conn.getMetaData().getDatabaseProductName()
+ .toLowerCase(Locale.ROOT);
+ if (!product.contains("h2") && !product.contains("mysql") &&
+ !product.contains("mariadb")) {
+ log.warn("Skip {} column migration for unsupported database " +
+ "product {}", table,
+ conn.getMetaData().getDatabaseProductName());
+ return;
+ }
+
+ for (String[] column : columns) {
+ String name = column[0];
+ boolean added = false;
+ if (!this.columnExists(conn, table, name)) {
+ try (Statement statement = conn.createStatement()) {
+ statement.execute(String.format(
+ "ALTER TABLE `%s` ADD COLUMN `%s` %s DEFAULT NULL",
+ table, name, column[1]));
+ }
+ added = true;
+ }
+ // Backfill unconditionally. The ALTER and the UPDATE commit
+ // separately, so a crash between them would leave the column
+ // present but permanently NULL: the next startup would see the
+ // column already exists and skip the backfill for good. The
+ // statement only touches NULL rows, so repeating it is a no-op.
+ if (column[2] != null) {
+ try (Statement statement = conn.createStatement()) {
+ statement.executeUpdate(String.format(
+ "UPDATE `%s` SET `%s` = %s WHERE `%s` IS NULL",
+ table, name, column[2], name));
+ }
+ }
+ if (added) {
+ log.info("Added {}.{} {}", table, name, column[1]);
+ }
+ }
+ }
+
+ private void addMissingIndexes(Connection conn) throws SQLException {
+ for (String[] definition : REQUIRED_INDEXES) {
+ String table = definition[0];
+ String index = definition[1];
+ if (!this.tableExists(conn, table) ||
+ this.indexExists(conn, table, index)) {
+ continue;
+ }
+ boolean columnsExist = true;
+ for (int i = 2; i < definition.length; i++) {
+ columnsExist &= this.columnExists(conn, table, definition[i]);
+ }
+ if (!columnsExist) {
+ continue;
+ }
+
+ StringBuilder columns = new StringBuilder();
+ for (int i = 2; i < definition.length; i++) {
+ if (columns.length() > 0) {
+ columns.append(", ");
+ }
+ columns.append('`').append(definition[i]).append('`');
+ }
+ try (Statement statement = conn.createStatement()) {
+ statement.execute(String.format(
+ "CREATE INDEX `%s` ON `%s`(%s)", index, table,
+ columns));
+ } catch (SQLException e) {
+ if (!this.indexExists(conn, table, index)) {
+ throw e;
+ }
+ log.info("Index {} on {} was added concurrently", index,
+ table);
+ continue;
+ }
+ log.info("Added index {} on {}", index, table);
+ }
+ }
+
+ private boolean indexExists(Connection conn, String table, String index)
+ throws SQLException {
+ DatabaseMetaData metaData = conn.getMetaData();
+ String[] tables = {table, table.toUpperCase(Locale.ROOT)};
+ for (String tableName : tables) {
+ try (ResultSet rs = metaData.getIndexInfo(conn.getCatalog(), null,
+ tableName, false,
+ false)) {
+ while (rs.next()) {
+ if (index.equalsIgnoreCase(rs.getString("INDEX_NAME"))) {
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
private void migrateFileMappingPath(Connection conn, int currentLength)
throws SQLException {
String sql = this.alterFileMappingPathSql(
@@ -187,6 +721,24 @@ public class DatabaseSchemaMigrator implements ApplicationRunner {
return false;
}
+ private boolean columnExists(Connection conn, String table, String column)
+ throws SQLException {
+ DatabaseMetaData metaData = conn.getMetaData();
+ String[] tables = {table, table.toUpperCase(Locale.ROOT)};
+ String[] columns = {column, column.toUpperCase(Locale.ROOT)};
+ for (String tableName : tables) {
+ for (String columnName : columns) {
+ try (ResultSet rs = metaData.getColumns(null, null, tableName,
+ columnName)) {
+ if (rs.next()) {
+ return true;
+ }
+ }
+ }
+ }
+ return false;
+ }
+
private int columnSize(Connection conn, String table, String column)
throws SQLException {
DatabaseMetaData metaData = conn.getMetaData();
@@ -217,4 +769,45 @@ public class DatabaseSchemaMigrator implements ApplicationRunner {
}
return null;
}
+
+ private static final class LegacyScope {
+
+ private final Integer connectionId;
+ private final String graphspace;
+ private final String graph;
+
+ private LegacyScope(Integer connectionId, String graphspace,
+ String graph) {
+ this.connectionId = connectionId;
+ this.graphspace = graphspace;
+ this.graph = graph;
+ }
+ }
+
+ private static final class LegacyCollectionCollision {
+
+ private final String graphspace;
+ private final String graph;
+ private final String name;
+ private final String type;
+
+ private LegacyCollectionCollision(String graphspace, String graph,
+ String name, String type) {
+ this.graphspace = graphspace;
+ this.graph = graph;
+ this.name = name;
+ this.type = type;
+ }
+ }
+
+ private static final class LegacyCollectionRow {
+
+ private final int id;
+ private final Integer connectionId;
+
+ private LegacyCollectionRow(int id, Integer connectionId) {
+ this.id = id;
+ this.connectionId = connectionId;
+ }
+ }
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/graph/GraphController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/graph/GraphController.java
index 1acb0d5b9..ea4d38611 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/graph/GraphController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/graph/GraphController.java
@@ -176,31 +176,41 @@ public class GraphController extends BaseController {
List<String> elementIds) {
HugeClient client = this.authClient(graphSpace, graph);
ArrayList<ElementEditHistory> list = new ArrayList<>();
- HashSet<String> set = new HashSet<>(elementIds);
- if ("VERTEX".equals(type)) {
- for (String vertexId : set) {
- vertexId = UriUtils.decode(vertexId, Constant.CHARSET);
- Vertex vertex = client.graph().getVertex(vertexId);
- String label = vertex.label();
- this.graphService.deleteVertex(client, vertexId);
- list.add(getEditEleHistory(graphSpace, graph, vertexId, label,
- vertex.properties().size(), OptionType.DELETE, ""));
+ HashSet<String> set = new HashSet<>();
+ for (String elementId : elementIds) {
+ set.add(UriUtils.decode(elementId, Constant.CHARSET));
+ }
+ try {
+ if ("VERTEX".equals(type)) {
+ for (String vertexId : set) {
+ Vertex vertex = client.graph().getVertex(vertexId);
+ String label = vertex.label();
+ this.graphService.deleteVertex(client, vertexId);
+ list.add(getEditEleHistory(graphSpace, graph, vertexId, label,
+ vertex.properties().size(), OptionType.DELETE, ""));
+ }
+ } else if ("EDGE".equals(type)) {
+ // Dedupe edge ids the same way as vertex ids, deleting the
+ // same edge twice fails after the first delete
+ for (String edgeId : set) {
+ Edge edge = client.graph().getEdge(edgeId);
+ String label = edge.label();
+ this.graphService.deleteEdge(client, edgeId);
+ list.add(getEditEleHistory(graphSpace, graph, edgeId, label,
+ edge.properties().size(),
+ OptionType.DELETE, ""));
+ }
+ } else {
+ throw new IllegalArgumentException(
+ "type must in [VERTEX, EDGE], but got '" + type + "'");
}
- } else if ("EDGE".equals(type)) {
- for (String edgeId : elementIds) {
- edgeId = UriUtils.decode(edgeId, Constant.CHARSET);
- Edge edge = client.graph().getEdge(edgeId);
- String label = edge.label();
- this.graphService.deleteEdge(client, edgeId);
- list.add(getEditEleHistory(graphSpace, graph, edgeId, label,
- edge.properties().size(),
- OptionType.DELETE, ""));
+ } finally {
+ // Persist the history of what was actually deleted even if the
+ // batch failed part way through
+ if (!list.isEmpty()) {
+ editEleHisService.add(list);
}
- } else {
- throw new IllegalArgumentException(
- "type must in [VERTEX, EDGE], but got '" + type + "'");
}
- editEleHisService.add(list);
}
@GetMapping("edgelabel/{label}")
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/ingest/IngestController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/ingest/IngestController.java
index b464063c4..7b7f97050 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/ingest/IngestController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/ingest/IngestController.java
@@ -83,6 +83,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
@@ -158,6 +159,7 @@ public class IngestController extends BaseController {
Long limit = this.config.get(HubbleOptions.UPLOAD_SINGLE_FILE_SIZE_LIMIT);
Ex.check(file.getSize() <= limit, "load.upload.file.exceed-single-size",
limit);
+ this.checkFileFormatValid(fileName);
Path uploadRoot = Paths.get(this.config.get(HubbleOptions.UPLOAD_FILE_LOCATION))
.toAbsolutePath()
@@ -504,6 +506,26 @@ public class IngestController extends BaseController {
// ===== Helpers =====
+ /**
+ * Same format-whitelist check as
+ * FileUploadController#checkFileValid
+ */
+ private void checkFileFormatValid(String fileName) {
+ String format = FilenameUtils.getExtension(fileName);
+ Ex.check(StringUtils.isNotBlank(format),
+ "load.upload.file.format.unsupported");
+ List<String> formatWhiteList = this.config.get(
+ HubbleOptions.UPLOAD_FILE_FORMAT_LIST);
+ String normalizedFormat = format.toLowerCase(Locale.ROOT);
+ boolean supported = formatWhiteList != null &&
+ formatWhiteList.stream()
+ .filter(StringUtils::isNotBlank)
+ .map(String::trim)
+ .map(item -> item.toLowerCase(Locale.ROOT))
+ .anyMatch(normalizedFormat::equals);
+ Ex.check(supported, "load.upload.file.format.unsupported");
+ }
+
private GraphConnection graphConnection(String graphSpace, String graph) {
GraphConnection connection = new GraphConnection();
connection.setCluster(config.get(HubbleOptions.PD_CLUSTER));
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileMappingController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileMappingController.java
index a9642e0ac..39674b59e 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileMappingController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileMappingController.java
@@ -19,17 +19,21 @@
package org.apache.hugegraph.controller.load;
import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.apache.hugegraph.common.Constant;
import org.apache.hugegraph.controller.BaseController;
import org.apache.hugegraph.driver.HugeClient;
import org.apache.hugegraph.entity.enums.JobStatus;
+import org.apache.hugegraph.entity.enums.LoadStatus;
import org.apache.hugegraph.entity.load.FileMapping;
import org.apache.hugegraph.entity.load.FileSetting;
import org.apache.hugegraph.entity.load.ElementMapping;
import org.apache.hugegraph.entity.load.JobManager;
import org.apache.hugegraph.entity.load.LoadParameter;
+import org.apache.hugegraph.entity.load.LoadTask;
import org.apache.hugegraph.entity.load.VertexMapping;
import org.apache.hugegraph.entity.load.EdgeMapping;
import org.apache.hugegraph.entity.schema.EdgeLabelEntity;
@@ -37,6 +41,7 @@ import org.apache.hugegraph.entity.schema.VertexLabelEntity;
import org.apache.hugegraph.exception.ExternalException;
import org.apache.hugegraph.service.load.FileMappingService;
import org.apache.hugegraph.service.load.JobManagerService;
+import org.apache.hugegraph.service.load.LoadTaskService;
import org.apache.hugegraph.service.schema.EdgeLabelService;
import org.apache.hugegraph.service.schema.VertexLabelService;
import org.apache.hugegraph.util.CollectionUtil;
@@ -54,6 +59,7 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -71,6 +77,8 @@ public class FileMappingController extends BaseController {
private FileMappingService service;
@Autowired
private JobManagerService jobService;
+ @Autowired
+ private LoadTaskService taskService;
@GetMapping
public IPage<FileMapping> list(@PathVariable("graphspace") String graphSpace,
@@ -108,9 +116,9 @@ public class FileMappingController extends BaseController {
if (mapping == null) {
throw new ExternalException("load.file-mapping.not-exist.id", id);
}
+ this.checkNoRunningTask(jobId, ImmutableSet.of(mapping.getId()));
- this.service.deleteDiskFile(mapping);
- this.service.remove(id);
+ this.jobService.deleteMappings(jobId, ImmutableList.of(mapping));
}
@DeleteMapping
@@ -119,9 +127,14 @@ public class FileMappingController extends BaseController {
@PathVariable("jobId") int jobId) {
List<FileMapping> mappings = this.service.listByJob(graphSpace, graph,
jobId);
+ Set<Integer> fileIds = new HashSet<>();
for (FileMapping mapping : mappings) {
- this.service.remove(mapping.getId());
+ fileIds.add(mapping.getId());
}
+ // Check them all upfront, don't delete part of the mappings then fail
+ this.checkNoRunningTask(jobId, fileIds);
+
+ this.jobService.deleteMappings(jobId, mappings);
}
@PostMapping("{id}/file-setting")
@@ -324,6 +337,27 @@ public class FileMappingController extends BaseController {
return jobEntity;
}
+ /**
+ * Keep source and progress files while a task can still run, resume, or
+ * retry from them.
+ */
+ private void checkNoRunningTask(int jobId, Set<Integer> fileIds) {
+ for (LoadTask task : this.taskService.taskListByJob(jobId)) {
+ LoadStatus status = task.getStatus();
+ boolean needsSource = status == LoadStatus.RUNNING ||
+ status == LoadStatus.PAUSED ||
+ status == LoadStatus.FAILED ||
+ status == LoadStatus.STOPPED;
+ if (needsSource &&
+ task.getFileId() != null &&
+ fileIds.contains(task.getFileId())) {
+ throw new ExternalException(
+ "load.file-mapping.delete.task-in-running",
+ task.getFileId());
+ }
+ }
+ }
+
private void checkVertexMappingValid(HugeClient client,
VertexMapping vertexMapping,
FileMapping fileMapping) {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileUploadController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileUploadController.java
index 8efac6f53..99c36da84 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileUploadController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/FileUploadController.java
@@ -92,15 +92,26 @@ public class FileUploadController extends BaseController {
Ex.check(CollectionUtil.allUnique(fileNames),
"load.upload.file.duplicate-name");
Map<String, String> tokens = new HashMap<>();
- for (String fileName : fileNames) {
- this.checkFileNameValid(fileName);
- String token = this.service.generateFileToken(fileName);
- Ex.check(!this.uploadingTokenLocks().containsKey(token),
- "load.upload.file.token.existed");
- this.uploadingTokenLocks().put(token, new ReentrantReadWriteLock());
- tokens.put(fileName, token);
+ Map<String, ReadWriteLock> reservations = new HashMap<>();
+ try {
+ for (String fileName : fileNames) {
+ this.checkFileNameValid(fileName);
+ String token = this.service.generateFileToken(fileName);
+ ReadWriteLock lock = new ReentrantReadWriteLock();
+ // Atomic reservation to avoid a containsKey-then-put race
+ ReadWriteLock previous = this.uploadingTokenLocks()
+ .putIfAbsent(token, lock);
+ Ex.check(previous == null, "load.upload.file.token.existed");
+ reservations.put(token, lock);
+ tokens.put(fileName, token);
+ }
+ return tokens;
+ } catch (RuntimeException e) {
+ reservations.forEach((token, lock) -> {
+ this.uploadingTokenLocks().remove(token, lock);
+ });
+ throw e;
}
- return tokens;
}
@PostMapping
@@ -198,15 +209,7 @@ public class FileUploadController extends BaseController {
mapping.setTotalLines(FileUtil.countLines(mapping.getPath()));
mapping.setTotalSize(actualFileSize);
- // Move to the directory corresponding to the file mapping Id
- String newPath = this.service.moveToNextLevelDir(mapping);
- // Update file mapping stored path
- mapping.setPath(newPath);
- this.service.update(mapping);
- // Update Job Manager size
- long jobSize = currentJob.getJobSize() + mapping.getTotalSize();
- currentJob.setJobSize(jobSize);
- this.jobService.update(currentJob);
+ this.finalizeUpload(mapping);
result.setId(mapping.getId());
// Remove uploading file token
this.uploadingTokenLocks().remove(token);
@@ -217,6 +220,33 @@ public class FileUploadController extends BaseController {
}
}
+ private void finalizeUpload(FileMapping mapping) {
+ String originalPath = mapping.getPath();
+ String movedPath = this.service.moveToNextLevelDir(mapping);
+ mapping.setPath(movedPath);
+ try {
+ this.jobService.completeUpload(mapping);
+ } catch (RuntimeException failure) {
+ mapping.setPath(originalPath);
+ mapping.setFileStatus(FileMappingStatus.UPLOADING);
+ try {
+ this.service.restoreMovedUpload(movedPath, originalPath);
+ } catch (RuntimeException compensationFailure) {
+ // Keep the database cleanup record pointed at the real file
+ // when restoring the original path is impossible.
+ mapping.setPath(movedPath);
+ mapping.setFileStatus(FileMappingStatus.FAILURE);
+ try {
+ this.service.update(mapping);
+ } catch (RuntimeException recordFailure) {
+ compensationFailure.addSuppressed(recordFailure);
+ }
+ failure.addSuppressed(compensationFailure);
+ }
+ throw failure;
+ }
+ }
+
@DeleteMapping
public Boolean delete(@PathVariable("graphspace") String graphSpace,
@PathVariable("graph") String graph,
@@ -239,12 +269,8 @@ public class FileUploadController extends BaseController {
lock.writeLock().lock();
}
try {
- this.service.deleteDiskFile(mapping);
- log.info("Prepare to remove file mapping {}", mapping.getId());
- this.service.remove(mapping.getId());
- long jobSize = jobEntity.getJobSize() - mapping.getTotalSize();
- jobEntity.setJobSize(jobSize);
- this.jobService.update(jobEntity);
+ this.jobService.deleteMappings(jobId,
+ Collections.singletonList(mapping));
if (lock != null) {
this.uploadingTokenLocks().remove(token);
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/JobManagerController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/JobManagerController.java
index e12146128..f3c58a3e5 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/JobManagerController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/JobManagerController.java
@@ -201,9 +201,15 @@ public class JobManagerController {
if (task.getStatus() == LoadStatus.FAILED) {
FileMapping mapping = this.fmService.get(graphSpace, graph, id,
fileId);
- reason = this.taskService.readLoadFailedReason(mapping);
+ if (mapping == null) {
+ // The file mapping may have been deleted while the task
+ // still references its file id
+ reason = LoadTaskService.NO_ERROR_FILE_REASON;
+ } else {
+ reason = this.taskService.readLoadFailedReason(mapping);
+ }
}
- reasonResult.setTaskId(task.getJobId());
+ reasonResult.setTaskId(task.getId());
reasonResult.setFileId(task.getFileId());
reasonResult.setFileName(task.getFileName());
reasonResult.setReason(reason);
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/LoadTaskController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/LoadTaskController.java
index d16cacc94..66f534c18 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/LoadTaskController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/load/LoadTaskController.java
@@ -137,6 +137,12 @@ public class LoadTaskController extends BaseController {
if (task == null) {
throw new ExternalException("load.task.not-exist.id", id);
}
+ /*
+ * Removing a task that is still importing would leave the loader
+ * thread writing to the graph with no task to report progress to
+ */
+ Ex.check(!task.getStatus().inRunning(),
+ "load.task.delete.task-in-running", id);
this.service.remove(id);
}
@@ -224,7 +230,7 @@ public class LoadTaskController extends BaseController {
Ex.check(this.service.get(graphSpace, graph, jobId, taskId) != null,
"load.task.not-exist.id", taskId);
Ex.check(jobEntity.getJobStatus() == JobStatus.LOADING,
- "load.task.pause.no-permission");
+ "load.task.resume.no-permission");
try {
return this.service.resume(taskId, this.getToken());
} finally {
@@ -244,7 +250,7 @@ public class LoadTaskController extends BaseController {
Ex.check(this.service.get(graphSpace, graph, jobId, taskId) != null,
"load.task.not-exist.id", taskId);
Ex.check(jobEntity.getJobStatus() == JobStatus.LOADING,
- "load.task.pause.no-permission");
+ "load.task.stop.no-permission");
try {
return this.service.stop(taskId);
} finally {
@@ -264,7 +270,7 @@ public class LoadTaskController extends BaseController {
Ex.check(this.service.get(graphSpace, graph, jobId, taskId) != null,
"load.task.not-exist.id", taskId);
Ex.check(jobEntity.getJobStatus() == JobStatus.LOADING,
- "load.task.pause.no-permission");
+ "load.task.retry.no-permission");
try {
return this.service.retry(taskId, this.getToken());
} finally {
@@ -288,7 +294,13 @@ public class LoadTaskController extends BaseController {
Integer fileId = task.getFileId();
FileMapping mapping = this.fmService.get(graphSpace, graph, jobId,
fileId);
- String reason = this.service.readLoadFailedReason(mapping);
+ /*
+ * The file mapping may have been deleted while the task still
+ * references its id, there is no error file to read in that case
+ */
+ String reason = mapping == null ?
+ LoadTaskService.NO_ERROR_FILE_REASON :
+ this.service.readLoadFailedReason(mapping);
return Response.builder()
.status(Constant.STATUS_OK)
.data(reason)
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/HStoreController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/HStoreController.java
index 66cddcf75..f071e8c6e 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/HStoreController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/HStoreController.java
@@ -27,6 +27,7 @@ import com.google.common.collect.ImmutableMap;
import lombok.extern.log4j.Log4j2;
import org.apache.hugegraph.service.space.HStoreService;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
@@ -59,7 +60,11 @@ public class HStoreController extends BaseController {
defaultValue = "10") int pageSize) {
HugeClient client = this.authClient(null, null);
- return hStoreService.listPage(client, pageNo, pageSize);
+ try {
+ return hStoreService.listPage(client, pageNo, pageSize);
+ } catch (ServerException e) {
+ throw mapHstoreFailure(e);
+ }
}
@GetMapping("nodes/{id}")
@@ -67,8 +72,13 @@ public class HStoreController extends BaseController {
HugeClient client = this.authClient(null, null);
E.checkNotNull(id, "id");
- List<HStoreNodeInfo.HStorePartitionInfo> partitions =
- client.hStoreManager().get(id).hStorePartitionInfoList();
+ List<HStoreNodeInfo.HStorePartitionInfo> partitions;
+ try {
+ partitions = client.hStoreManager().get(id)
+ .hStorePartitionInfoList();
+ } catch (ServerException e) {
+ throw mapHstoreFailure(e, id);
+ }
return ImmutableMap.of("shards", partitions);
}
@@ -91,8 +101,7 @@ public class HStoreController extends BaseController {
try {
status = client.hStoreManager().status();
} catch (ServerException e) {
- throw new ServerCapabilityUnavailableException(
- "server.capability.hstore-status.unavailable", e);
+ throw mapHstoreFailure(e);
}
return ImmutableMap.of("status", status);
@@ -104,7 +113,11 @@ public class HStoreController extends BaseController {
@GetMapping("split")
public void triggerSplit() {
HugeClient client = this.authClient(null, null);
- client.hStoreManager().startSplit();
+ try {
+ client.hStoreManager().startSplit();
+ } catch (ServerException e) {
+ throw mapHstoreFailure(e);
+ }
}
/**
@@ -124,6 +137,11 @@ public class HStoreController extends BaseController {
for (String nodeId: nodes) {
try {
client.hStoreManager().nodeStartup(nodeId);
+ } catch (ServerException e) {
+ log.warn("Startup of HStore node {} failed after nodes {} " +
+ "succeeded", nodeId, successNodes);
+ throw mapHstoreFailure(e, nodeId,
+ ImmutableList.copyOf(successNodes));
} catch (RuntimeException e) {
log.info("startup hstore node({}) success", successNodes);
log.warn("startup hstore node({}) error", nodeId, e);
@@ -153,6 +171,11 @@ public class HStoreController extends BaseController {
for (String nodeId: nodes) {
try {
client.hStoreManager().nodeShutdown(nodeId);
+ } catch (ServerException e) {
+ log.warn("Shutdown of HStore node {} failed after nodes {} " +
+ "succeeded", nodeId, successNodes);
+ throw mapHstoreFailure(e, nodeId,
+ ImmutableList.copyOf(successNodes));
} catch (RuntimeException e) {
log.info("shutdown hstore node({}) success", successNodes);
log.warn("shutdown hstore node({}) error", nodeId, e);
@@ -164,4 +187,17 @@ public class HStoreController extends BaseController {
successNodes.add(nodeId);
}
}
+
+ static RuntimeException mapHstoreFailure(ServerException exception,
+ Object... context) {
+ int status = exception.status();
+ if (status == HttpStatus.NOT_FOUND.value() ||
+ status == HttpStatus.METHOD_NOT_ALLOWED.value() ||
+ status == HttpStatus.NOT_IMPLEMENTED.value()) {
+ return new ServerCapabilityUnavailableException(
+ "server.capability.hstore-status.unavailable", exception,
+ context);
+ }
+ return exception;
+ }
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/LogController.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/LogController.java
index 4ade014cb..60393f8e2 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/LogController.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/op/LogController.java
@@ -37,6 +37,7 @@ import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
+import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.List;
@@ -79,8 +80,13 @@ public class LogController extends BaseController {
@PostMapping("export")
public void export(HttpServletResponse response,
@RequestBody LogService.LogReq logReq) {
- String fileName = String.format("log.txt", logReq.startDatetime,
- logReq.endDatetime);
+ SimpleDateFormat dateFormat =
+ new SimpleDateFormat("yyyyMMddHHmmss");
+ String start = logReq.startDatetime == null ?
+ "begin" : dateFormat.format(logReq.startDatetime);
+ String end = logReq.endDatetime == null ?
+ "end" : dateFormat.format(logReq.endDatetime);
+ String fileName = String.format("log_%s_%s.txt", start, end);
response.setCharacterEncoding("UTF-8");
response.setContentType("application/octet-stream");
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/load/LoadTask.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/load/LoadTask.java
index 30acc4c56..ce6fd04af 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/load/LoadTask.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/load/LoadTask.java
@@ -258,10 +258,14 @@ public class LoadTask implements Runnable {
if (this.fileTotalLines == null || this.fileTotalLines == 0) {
return 0;
}
- Ex.check(this.fileTotalLines >= this.fileReadLines,
- "The file total lines must be >= read lines, " +
- "but got total lines %s, read lines %s",
- this.fileTotalLines, this.fileReadLines);
+ // Clamp instead of throwing: this getter runs during JSON
+ // serialization, an exception here would fail the whole response
+ if (this.fileReadLines == null) {
+ return 0;
+ }
+ if (this.fileReadLines >= this.fileTotalLines) {
+ return 100;
+ }
float actualProgress = (float) this.fileReadLines / this.fileTotalLines;
return ((int) (actualProgress * 10000)) / 100.0F;
}
@@ -279,7 +283,7 @@ public class LoadTask implements Runnable {
@JsonProperty("load_rate")
public String getLoadRate() {
- long readLines = this.fileReadLines;
+ long readLines = this.fileReadLines == null ? 0L : this.fileReadLines;
long duration = this.getDuration();
float rate;
if (readLines == 0L || duration == 0L) {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ApplicationInfo.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ApplicationInfo.java
index 23b8bafb4..974b43649 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ApplicationInfo.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ApplicationInfo.java
@@ -21,14 +21,18 @@ package org.apache.hugegraph.entity.query;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
+import lombok.NoArgsConstructor;
import org.apache.hugegraph.annotation.MergeProperty;
import org.apache.hugegraph.common.AppType;
import org.apache.hugegraph.common.Mergeable;
import org.apache.hugegraph.util.JsonUtil;
@Data
+@NoArgsConstructor
+@AllArgsConstructor
@Builder
@TableName("app_info")
public class ApplicationInfo implements Mergeable {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ElementEditHistory.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ElementEditHistory.java
index 811fc160d..a69a295c4 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ElementEditHistory.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/entity/query/ElementEditHistory.java
@@ -32,8 +32,10 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Builder;
import lombok.Data;
+import lombok.NoArgsConstructor;
@Data
+@NoArgsConstructor
@Builder
@TableName("edit_history")
public class ElementEditHistory implements Identifiable, Mergeable {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/handler/ExceptionAdvisor.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/handler/ExceptionAdvisor.java
index c64f42786..2a23c422c 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/handler/ExceptionAdvisor.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/handler/ExceptionAdvisor.java
@@ -370,12 +370,15 @@ public class ExceptionAdvisor {
String origin = result.get("message").toString();
List<String> attach =
HubbleUtil.uncheckedCast(result.get("attach"));
+ Object[] args = attach != null ? attach.toArray() :
+ new Object[0];
message = ErrorCodeMessage.getErrorMessage(code, origin,
- attach.toArray());
+ args);
}
} catch (Exception e) {
+ // Fall back to the original upstream message, never throw
+ // from inside the exception handler
log.error("hubble.error_code_parse_failed");
- throw new RuntimeException("failed_to_handle_error_code");
}
}
return message;
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/FileMappingMapper.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/FileMappingMapper.java
index bfb4cfd87..69d537755 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/FileMappingMapper.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/FileMappingMapper.java
@@ -19,6 +19,8 @@
package org.apache.hugegraph.mapper.load;
import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Update;
import org.springframework.stereotype.Component;
import org.apache.hugegraph.entity.load.FileMapping;
@@ -27,4 +29,9 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
@Mapper
@Component
public interface FileMappingMapper extends BaseMapper<FileMapping> {
+
+ @Update("UPDATE file_mapping SET job_id = -id " +
+ "WHERE id = #{mappingId} AND job_id = #{jobId}")
+ int detachFromJob(@Param("mappingId") int mappingId,
+ @Param("jobId") int jobId);
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/JobManagerMapper.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/JobManagerMapper.java
index f1eb255b5..628149c82 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/JobManagerMapper.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/load/JobManagerMapper.java
@@ -21,6 +21,7 @@ package org.apache.hugegraph.mapper.load;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
import org.springframework.stereotype.Component;
import org.apache.hugegraph.entity.load.JobManager;
@@ -31,9 +32,18 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
@Component
public interface JobManagerMapper extends BaseMapper<JobManager> {
- @Select("SELECT ISNULL(SUM(f.total_size),0) as total_size, " +
- "ISNULL(SUM(l.duration),0) as duration " +
+ @Select("SELECT IFNULL(SUM(f.total_size),0) as total_size, " +
+ "IFNULL(SUM(l.last_duration + l.curr_duration),0) as duration " +
"FROM `load_task` as l LEFT JOIN `file_mapping` as f " +
"ON l.file_id=f.id WHERE l.job_id = #{job_id}")
JobManagerItem computeSizeDuration(@Param("job_id") int jobId);
+
+ @Update("UPDATE `job_manager` SET `job_size` = `job_size` + #{size} " +
+ "WHERE `id` = #{id}")
+ int increaseJobSize(@Param("id") int id, @Param("size") long size);
+
+ @Update("UPDATE `job_manager` SET `job_size` = CASE " +
+ "WHEN `job_size` >= #{size} THEN `job_size` - #{size} " +
+ "ELSE 0 END WHERE `id` = #{id}")
+ int decreaseJobSize(@Param("id") int id, @Param("size") long size);
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/EditElementHistoryMapper.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/EditElementHistoryMapper.java
index 53badfe49..991e46409 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/EditElementHistoryMapper.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/EditElementHistoryMapper.java
@@ -32,8 +32,12 @@ import java.util.List;
@Component
public interface EditElementHistoryMapper extends
BaseMapper<ElementEditHistory> {
- @Select("SELECT * FROM `edit_history` LIMIT #{limit}")
- List<ElementEditHistory> queryByLimit(@Param("limit") int limit);
+ @Select("SELECT * FROM `edit_history` WHERE graphspace = #{graphspace} " +
+ "AND graph = #{graph} ORDER BY option_time DESC LIMIT #{limit}")
+ List<ElementEditHistory> queryByLimit(
+ @Param("graphspace") String graphspace,
+ @Param("graph") String graph,
+ @Param("limit") int limit);
@Select("SELECT * FROM `edit_history` WHERE graphspace = #{graphspace} " +
"AND graph = #{graph} AND element_id = #{elementId}")
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/ExecuteHistoryMapper.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/ExecuteHistoryMapper.java
index 4d62ad362..543f4802f 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/ExecuteHistoryMapper.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/mapper/query/ExecuteHistoryMapper.java
@@ -32,16 +32,53 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
@Component
public interface ExecuteHistoryMapper extends BaseMapper<ExecuteHistory> {
- // 删除超出限制的记录
+ // 单条 DELETE ... IN (...) 允许携带的最大 id 数, 避免超出数据库
+ // 的报文大小与参数个数上限
+ int DELETE_BATCH_SIZE = 500;
+
+ // 删除超出限制的记录, 按 (graphspace, graph) 分别保留最新的 limit 条
default void deleteExceedLimit(int limit) {
- List<Long> ids = findIdsToDelete(limit, limit);
- if (!ids.isEmpty()) {
- deleteBatchIds(ids);
+ for (ExecuteHistory scope : findScopes()) {
+ while (true) {
+ int excess = countScope(scope.getGraphspace(),
+ scope.getGraph()) - limit;
+ if (excess <= 0) {
+ break;
+ }
+ int batchSize = Math.min(excess, DELETE_BATCH_SIZE);
+ List<Long> ids = findIdsOldestFirst(scope.getGraphspace(),
+ scope.getGraph(),
+ batchSize);
+ if (ids.isEmpty()) {
+ break;
+ }
+ // Another scheduler may have deleted the same candidates.
+ // Recheck after selection to avoid consuming retained rows.
+ excess = countScope(scope.getGraphspace(), scope.getGraph()) -
+ limit;
+ if (excess <= 0) {
+ break;
+ }
+ deleteBatchIds(ids.subList(0, Math.min(excess, ids.size())));
+ }
}
}
- // 查询满足条件的 id 列表
- @Select("SELECT id FROM `execute_history` ORDER BY `create_time` DESC " +
- "LIMIT #{limit} OFFSET #{offset}")
- List<Long> findIdsToDelete(@Param("limit") int limit, @Param("offset") int offset);
+ // 查询所有出现过的 (graphspace, graph) 组合
+ @Select("SELECT `graphspace`, `graph` FROM `execute_history` " +
+ "GROUP BY `graphspace`, `graph`")
+ List<ExecuteHistory> findScopes();
+
+ @Select("SELECT COUNT(*) FROM `execute_history` " +
+ "WHERE `graphspace` = #{graphspace} AND `graph` = #{graph}")
+ int countScope(@Param("graphspace") String graphspace,
+ @Param("graph") String graph);
+
+ @Select("SELECT `id` FROM `execute_history` " +
+ "WHERE `graphspace` = #{graphspace} AND `graph` = #{graph} " +
+ "ORDER BY `create_time` ASC, `id` ASC " +
+ "LIMIT #{batchSize}")
+ List<Long> findIdsOldestFirst(@Param("graphspace") String graphspace,
+ @Param("graph") String graph,
+ @Param("batchSize") int batchSize);
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/graphs/GraphsService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/graphs/GraphsService.java
index 6361aed2c..d2eb78e8d 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/graphs/GraphsService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/graphs/GraphsService.java
@@ -71,6 +71,7 @@ import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -454,7 +455,8 @@ public class GraphsService {
String graphSpace,
String graph) {
GraphStatisticsEntity result;
- this.graphStatistics.clear();
+ // Only invalidate the cache entry of the current graph
+ this.graphStatistics.remove(getStatisticsKey(graphSpace, graph));
result = postSmallStatistics(client, graphSpace, graph);
return result;
@@ -464,30 +466,39 @@ public class GraphsService {
HugeClient client,
String graphSpace,
String graph) {
- if (isBigGraph(pdClient, graphSpace, graph)) {
- GremlinQuery query = new GremlinQuery(GREMLIN_STATISTICS_VERTEX);
- long vid = executeAsyncTask(client, graphSpace, graph, query);
+ if (!isBigGraph(pdClient, graphSpace, graph)) {
+ return;
+ }
+ String graphKey = getStatisticsKey(graphSpace, graph);
+ GremlinQuery query = new GremlinQuery(GREMLIN_STATISTICS_VERTEX);
+ long vid = executeAsyncTask(client, graphSpace, graph, query);
+ long eid;
+ try {
query.setContent(GREMLIN_STATISTICS_EDGE);
- long eid = executeAsyncTask(client, graphSpace, graph, query);
- String idPair = String.valueOf(vid) + "-" + String.valueOf(eid);
- String graphKey = getStatisticsKey(graphSpace, graph);
- if (this.graphStatistics.containsKey(graphKey)) {
- Map<String, Object> graphCache =
- this.graphStatistics.get(graphKey);
- if (graphCache.get(RUNNING_TASKS) != null) {
- List<String> idPairs =
- HubbleUtil.uncheckedCast(graphCache.get(RUNNING_TASKS));
- idPairs.add(idPair);
- return;
- }
+ eid = executeAsyncTask(client, graphSpace, graph, query);
+ } catch (RuntimeException e) {
+ try {
+ client.task().cancel(vid);
+ } catch (RuntimeException cleanup) {
+ log.warn("Failed to cancel partial statistics task {}: {}",
+ vid, cleanup.getMessage());
}
- List<String> idPairs = new ArrayList<>();
- idPairs.add(idPair);
- Map<String, Object> graphCache = new HashMap<>(2);
- graphCache.put(RUNNING_TASKS, idPairs);
- graphCache.put(STATISTICS, GraphStatisticsEntity.emptyEntity());
- this.graphStatistics.put(graphKey, graphCache);
+ throw e;
}
+ String idPair = String.valueOf(vid) + "-" + String.valueOf(eid);
+ this.graphStatistics.compute(graphKey, (key, graphCache) -> {
+ if (graphCache == null) {
+ graphCache = new ConcurrentHashMap<>(2);
+ graphCache.put(RUNNING_TASKS,
+ new CopyOnWriteArrayList<String>());
+ graphCache.put(STATISTICS,
+ GraphStatisticsEntity.emptyEntity());
+ }
+ List<String> idPairs =
+ HubbleUtil.uncheckedCast(graphCache.get(RUNNING_TASKS));
+ idPairs.add(idPair);
+ return graphCache;
+ });
}
public GraphStatisticsEntity getLastStatistics(HugeClient client,
@@ -526,6 +537,12 @@ public class GraphsService {
String[] idVE = idPair.split("-");
Task taskV = taskMap.get(idVE[0]);
Task taskE = taskMap.get(idVE[1]);
+ if (taskV == null || taskE == null) {
+ // The server no longer knows these tasks. Drop the pair,
+ // otherwise it is re-queried on every call and never drains.
+ removeIds.add(idPair);
+ continue;
+ }
boolean success = taskV.success() && taskE.success();
if (removable(taskV) || removable(taskE) || success) {
removeIds.add(idPair);
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/DatasourceService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/DatasourceService.java
index 62dc3a4f1..ac6d6627e 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/DatasourceService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/DatasourceService.java
@@ -25,6 +25,7 @@ import org.apache.hugegraph.entity.load.Datasource;
import org.apache.hugegraph.exception.InternalException;
import org.apache.hugegraph.mapper.load.DatasourceMapper;
import org.apache.hugegraph.util.HubbleUtil;
+import org.apache.hugegraph.util.PageUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
@@ -46,7 +47,8 @@ public class DatasourceService {
wrapper.like("datasource_name", query);
}
wrapper.orderByDesc("create_time");
- return this.mapper.selectPage(new Page<>(pageNo, pageSize), wrapper);
+ return this.mapper.selectPage(
+ new Page<>(pageNo, PageUtil.boundedSize(pageSize)), wrapper);
}
@Transactional(isolation = Isolation.READ_COMMITTED)
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/FileMappingService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/FileMappingService.java
index 06495a763..0e531df83 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/FileMappingService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/FileMappingService.java
@@ -27,8 +27,10 @@ import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -66,6 +68,7 @@ import org.apache.hugegraph.mapper.load.JobManagerMapper;
import org.apache.hugegraph.options.HubbleOptions;
import org.apache.hugegraph.util.Ex;
import org.apache.hugegraph.util.HubbleUtil;
+import org.apache.hugegraph.util.PageUtil;
import org.apache.hugegraph.util.StringUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -82,6 +85,11 @@ public class FileMappingService {
public static final String JOB_PREIFX = "job-";
public static final String FILE_PREIFX = "file-mapping-";
+ private static final String TOKEN_SUFFIX_CHARS =
+ "0123456789abcdefghijklmnopqrstuvwxyz";
+ private static final int TOKEN_SUFFIX_LENGTH = 4;
+ private static final SecureRandom TOKEN_RANDOM = new SecureRandom();
+
@Autowired
private HugeConfig config;
@Autowired
@@ -149,7 +157,8 @@ public class FileMappingService {
query.eq("job_id", jobId);
query.eq("file_status", FileMappingStatus.COMPLETED.getValue());
query.orderByDesc("create_time");
- Page<FileMapping> page = new Page<>(pageNo, pageSize);
+ Page<FileMapping> page = new Page<>(pageNo,
+ PageUtil.boundedSize(pageSize));
return this.mapper.selectPage(page, query);
}
@@ -174,9 +183,32 @@ public class FileMappingService {
}
}
+ @Transactional(isolation = Isolation.READ_COMMITTED)
+ public void detachFromJob(int mappingId, int jobId) {
+ if (this.mapper.detachFromJob(mappingId, jobId) != 1) {
+ throw new InternalException("entity.update.failed", mappingId);
+ }
+ }
+
public String generateFileToken(String fileName) {
+ /*
+ * Append a short random suffix so that tokens cannot be guessed
+ * from the file name and upload time alone. Note that
+ * checkFileNameMatchToken() only checks the md5 prefix, so the
+ * token still passes that check.
+ */
return this.fileTokenPrefix(fileName) +
- HubbleUtil.nowTime().getEpochSecond();
+ HubbleUtil.nowTime().getEpochSecond() + "-" +
+ randomTokenSuffix();
+ }
+
+ private static String randomTokenSuffix() {
+ StringBuilder suffix = new StringBuilder(TOKEN_SUFFIX_LENGTH);
+ for (int i = 0; i < TOKEN_SUFFIX_LENGTH; i++) {
+ int index = TOKEN_RANDOM.nextInt(TOKEN_SUFFIX_CHARS.length());
+ suffix.append(TOKEN_SUFFIX_CHARS.charAt(index));
+ }
+ return suffix.toString();
}
public FileUploadResult uploadFile(MultipartFile srcFile, String fileName,
@@ -372,6 +404,35 @@ public class FileMappingService {
return Paths.get(destPath, currFile.getName()).toString();
}
+ public void restoreMovedUpload(String movedPath, String originalPath) {
+ File movedFile = this.requirePathUnderUploadRoot(movedPath);
+ File originalFile = this.requirePathUnderUploadRoot(originalPath);
+ if (!movedFile.exists()) {
+ if (originalFile.exists()) {
+ return;
+ }
+ throw new InternalException(
+ "Failed to restore upload file after database rollback");
+ }
+ try {
+ originalFile = this.requirePathUnderUploadRoot(originalPath);
+ FileUtils.moveFile(movedFile, originalFile);
+ } catch (IOException e) {
+ throw new InternalException(
+ "Failed to restore upload file after database rollback",
+ e);
+ }
+
+ File movedDirectory = movedFile.getParentFile();
+ String[] children = movedDirectory == null ? null :
+ movedDirectory.list();
+ if (children != null && children.length == 0 &&
+ !movedDirectory.delete()) {
+ log.warn("Failed to remove empty upload directory {}",
+ movedDirectory);
+ }
+ }
+
public void deleteDiskFile(FileMapping mapping) {
File file = this.requirePathUnderUploadRoot(mapping.getPath());
if (file.isDirectory()) {
@@ -403,7 +464,8 @@ public class FileMappingService {
@Scheduled(fixedRate = 10 * 60 * 1000)
public void deleteUnfinishedFile() {
QueryWrapper<FileMapping> query = Wrappers.query();
- query.in("file_status", FileMappingStatus.UPLOADING.getValue());
+ query.in("file_status", FileMappingStatus.UPLOADING.getValue(),
+ FileMappingStatus.FAILURE.getValue());
List<FileMapping> mappings = this.mapper.selectList(query);
long threshold = this.config.get(
HubbleOptions.UPLOAD_FILE_MAX_TIME_CONSUMING) * 1000;
@@ -501,25 +563,38 @@ public class FileMappingService {
}
private Path normalizePath(File file) {
- Path path = file.toPath();
+ Path path = file.toPath().toAbsolutePath().normalize();
try {
if (file.exists()) {
return path.toRealPath();
}
+ Path existing = path.getParent();
+ while (existing != null && !Files.exists(existing)) {
+ existing = existing.getParent();
+ }
+ if (existing != null) {
+ Path resolvedParent = existing.toRealPath();
+ return resolvedParent.resolve(existing.relativize(path))
+ .normalize();
+ }
} catch (IOException e) {
throw new InternalException("Failed to resolve upload path '%s'",
e, file);
}
- return path.toAbsolutePath().normalize();
+ return path;
}
private void tryDeleteUnfinishedMapping(FileMapping mapping) {
String filePath = mapping.getPath();
try {
- FileUtils.forceDelete(this.requirePathUnderUploadRoot(filePath));
+ File file = this.requirePathUnderUploadRoot(filePath);
+ if (file.exists()) {
+ FileUtils.forceDelete(file);
+ }
} catch (IOException e) {
log.warn("Failed to delete expired uploading file {}",
filePath, e);
+ return;
} catch (RuntimeException e) {
log.warn("Skip deleting expired uploading file {} because the " +
"path is invalid", filePath, e);
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/JobManagerService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/JobManagerService.java
index b724c8e9f..ecbf666ed 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/JobManagerService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/JobManagerService.java
@@ -31,6 +31,7 @@ import lombok.extern.log4j.Log4j2;
import org.apache.hugegraph.driver.HugeClient;
import org.apache.hugegraph.entity.GraphConnection;
+import org.apache.hugegraph.entity.enums.FileMappingStatus;
import org.apache.hugegraph.entity.enums.JobStatus;
import org.apache.hugegraph.entity.enums.LoadStatus;
import org.apache.hugegraph.entity.load.FileMapping;
@@ -40,6 +41,7 @@ import org.apache.hugegraph.exception.ExternalException;
import org.apache.hugegraph.exception.InternalException;
import org.apache.hugegraph.mapper.load.JobManagerMapper;
import org.apache.hugegraph.util.HubbleUtil;
+import org.apache.hugegraph.util.PageUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
@@ -100,7 +102,8 @@ public class JobManagerService {
query.like("job_name", content);
}
query.orderByDesc("create_time");
- Page<JobManager> page = new Page<>(pageNo, pageSize);
+ Page<JobManager> page = new Page<>(pageNo,
+ PageUtil.boundedSize(pageSize));
IPage<JobManager> list = this.mapper.selectPage(page, query);
list.getRecords().forEach(task -> {
this.refreshStatus(task);
@@ -122,9 +125,8 @@ public class JobManagerService {
query.like("job_name", content);
}
query.orderByDesc("create_time");
- IPage<JobManager> list = this.mapper.selectPage(new Page<>(pageNo,
- pageSize),
- query);
+ IPage<JobManager> list = this.mapper.selectPage(
+ new Page<>(pageNo, PageUtil.boundedSize(pageSize)), query);
list.getRecords().forEach(this::refreshStatus);
return list;
}
@@ -180,6 +182,41 @@ public class JobManagerService {
}
}
+ @Transactional(isolation = Isolation.READ_COMMITTED)
+ public void completeUpload(FileMapping mapping) {
+ Integer jobId = mapping.getJobId();
+ if (jobId == null) {
+ throw new InternalException("File mapping has no job id");
+ }
+ this.fileMappingService.update(mapping);
+ if (this.mapper.increaseJobSize(jobId, mapping.getTotalSize()) != 1) {
+ throw new InternalException("entity.update.failed", jobId);
+ }
+ }
+
+ @Transactional(isolation = Isolation.READ_COMMITTED)
+ public void deleteMappings(int jobId, List<FileMapping> mappings) {
+ if (mappings.isEmpty()) {
+ return;
+ }
+ long releasedSize = 0L;
+ for (FileMapping mapping : mappings) {
+ if (mapping.getJobId() == null || mapping.getJobId() != jobId) {
+ throw new InternalException("File mapping does not belong to " +
+ "job %s", jobId);
+ }
+ if (mapping.getFileStatus() == FileMappingStatus.COMPLETED) {
+ releasedSize = Math.addExact(releasedSize,
+ mapping.getTotalSize());
+ }
+ this.fileMappingService.detachFromJob(mapping.getId(), jobId);
+ }
+ if (this.mapper.decreaseJobSize(jobId, releasedSize) != 1) {
+ throw new InternalException("entity.update.failed", jobId);
+ }
+ this.deleteDiskFilesAfterCommit(mappings);
+ }
+
@Transactional(isolation = Isolation.READ_COMMITTED)
public void remove(int id) {
if (this.mapper.deleteById(id) != 1) {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/LoadTaskService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/LoadTaskService.java
index a07a09bd2..32aef50da 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/LoadTaskService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/load/LoadTaskService.java
@@ -59,6 +59,7 @@ import org.apache.hugegraph.service.schema.EdgeLabelService;
import org.apache.hugegraph.service.schema.VertexLabelService;
import org.apache.hugegraph.util.Ex;
import org.apache.hugegraph.util.HubbleUtil;
+import org.apache.hugegraph.util.PageUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
@@ -76,6 +77,8 @@ import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.file.Paths;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -88,6 +91,10 @@ import org.apache.commons.lang3.StringUtils;
@Service
public class LoadTaskService {
+ public static final String NO_ERROR_FILE_REASON =
+ "For some reason, the error file was not generated. " +
+ "Please check the log for details";
+
@Autowired
private LoadTaskMapper mapper;
@Autowired
@@ -134,7 +141,8 @@ public class LoadTaskService {
query.eq("graph", graph);
query.eq("job_id", jobId);
query.orderByDesc("create_time");
- Page<LoadTask> page = new Page<>(pageNo, pageSize);
+ Page<LoadTask> page = new Page<>(pageNo,
+ PageUtil.boundedSize(pageSize));
return this.mapper.selectPage(page, query);
}
@@ -302,13 +310,24 @@ public class LoadTaskService {
}
public LoadTask pause(int taskId) {
+ /*
+ * The in-memory container is not repopulated on startup, so a task
+ * that was RUNNING before a restart is only visible in the database
+ */
LoadTask task = this.runningTaskContainer.get(taskId);
+ boolean loading = task != null;
+ if (!loading) {
+ task = this.get(taskId);
+ }
+ Ex.check(task != null, "load.task.not-exist.id", taskId);
Ex.check(task.getStatus() == LoadStatus.RUNNING,
"Can only pause the RUNNING task");
// Mark status as paused, should set before context.stopLoading()
task.setStatus(LoadStatus.PAUSED);
- // Let HugeGraphLoader stop
- task.stop();
+ if (loading) {
+ // Let HugeGraphLoader stop
+ task.stop();
+ }
task.lock();
try {
@@ -342,15 +361,22 @@ public class LoadTaskService {
public LoadTask stop(int taskId) {
LoadTask task = this.runningTaskContainer.get(taskId);
- if (task == null) {
+ boolean loading = task != null;
+ if (!loading) {
task = this.get(taskId);
}
+ Ex.check(task != null, "load.task.not-exist.id", taskId);
LoadStatus status = task.getStatus();
Ex.check(status == LoadStatus.RUNNING || status == LoadStatus.PAUSED,
"Can only stop the RUNNING or PAUSED task");
// Mark status as stopped
task.setStatus(LoadStatus.STOPPED);
- if (status == LoadStatus.RUNNING) {
+ /*
+ * Only a task taken from the in-memory container owns a loader, one
+ * restored from the database has nothing running to interrupt, so
+ * just persist the terminal status for it
+ */
+ if (loading && status == LoadStatus.RUNNING) {
task.stop();
}
@@ -387,6 +413,9 @@ public class LoadTaskService {
}
public String readLoadFailedReason(FileMapping mapping) {
+ if (mapping == null) {
+ return NO_ERROR_FILE_REASON;
+ }
String path = mapping.getPath();
File parentDir = FileUtils.getFile(path).getParentFile();
File failureDataDir = FileUtils.getFile(parentDir, "mapping",
@@ -395,21 +424,28 @@ public class LoadTaskService {
File[] errorFiles = failureDataDir.listFiles((dir, name) -> {
return name.endsWith("error");
});
- if (errorFiles == null) {
- return "For some reason, the error file was not generated. " +
- "Please check the log for details";
- }
- Ex.check(errorFiles.length == 1,
- "There should exist only one error file, actual is %s",
- errorFiles.length);
- File errorFile = errorFiles[0];
- try {
- return FileUtils.readFileToString(errorFile,
- Charset.defaultCharset());
- } catch (IOException e) {
- throw new InternalException("Failed to read error file %s",
- e, errorFile);
+ if (errorFiles == null || errorFiles.length == 0) {
+ return NO_ERROR_FILE_REASON;
+ }
+ /*
+ * A single run can legitimately produce both a parse error file and
+ * an insert error file, report the contents of all of them
+ */
+ Arrays.sort(errorFiles, Comparator.comparing(File::getName));
+ StringBuilder reason = new StringBuilder();
+ for (File errorFile : errorFiles) {
+ if (reason.length() > 0) {
+ reason.append(System.lineSeparator());
+ }
+ try {
+ reason.append(FileUtils.readFileToString(
+ errorFile, Charset.defaultCharset()));
+ } catch (IOException e) {
+ throw new InternalException("Failed to read error file %s",
+ e, errorFile);
+ }
}
+ return reason.toString();
}
public void pauseAllTasks() {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/ESService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/ESService.java
index fe50678a6..b2004cb81 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/ESService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/ESService.java
@@ -44,8 +44,11 @@ public abstract class ESService {
@Autowired
private HugeConfig config;
- public static final String[] LEVELS = new String[]{"TRACE", "OFF",
- "FATAL", "ERROR", "WARN", "INFO", "DEBUG"};
+ // Ordered from the most severe to the least severe, LogService filters
+ // by Arrays.copyOfRange(LEVELS, 0, index + 1), so the prefix of this
+ // array must be exactly the levels at least as severe as the selected one
+ public static final String[] LEVELS = new String[]{"OFF", "FATAL",
+ "ERROR", "WARN", "INFO", "DEBUG", "TRACE"};
public static volatile ElasticsearchClient elasticsearchClient;
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/LogService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/LogService.java
index 15476a679..6c481d5c3 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/LogService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/op/LogService.java
@@ -146,8 +146,10 @@ public class LogService extends ESService {
HubbleUtil.uncheckedCast(hit.source())));
}
- int resultCount = (int) (search.hits().total().value());
- if (resultCount < batchSize) {
+ // Break on the size of the returned batch, not on the total
+ // number of matches: total() may be null and is unrelated to
+ // whether another page exists
+ if (search.hits().hits().size() < batchSize) {
break;
}
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/EditElementHistoryService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/EditElementHistoryService.java
index 34c5a9fc6..addbc5814 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/EditElementHistoryService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/EditElementHistoryService.java
@@ -43,8 +43,9 @@ public class EditElementHistoryService {
@Autowired
private EditElementHistoryMapper mapper;
- public List<ElementEditHistory> queryByLimit(int limit) {
- return mapper.queryByLimit(limit);
+ public List<ElementEditHistory> queryByLimit(String graphspace,
+ String graph, int limit) {
+ return mapper.queryByLimit(graphspace, graph, limit);
}
public int add(ElementEditHistory history) {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/ExecuteHistoryService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/ExecuteHistoryService.java
index 5239967f6..a56e85188 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/ExecuteHistoryService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/ExecuteHistoryService.java
@@ -40,6 +40,7 @@ import org.apache.hugegraph.mapper.query.ExecuteHistoryMapper;
import org.apache.hugegraph.options.HubbleOptions;
import org.apache.hugegraph.structure.Task;
import org.apache.hugegraph.util.HubbleUtil;
+import org.apache.hugegraph.util.PageUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
@@ -77,7 +78,8 @@ public class ExecuteHistoryService {
if (text2Gremlin) {
query.apply("LENGTH(text) > 0");
}
- Page<ExecuteHistory> page = new Page<>(current, pageSize);
+ Page<ExecuteHistory> page = new Page<>(current,
+ PageUtil.boundedSize(pageSize));
IPage<ExecuteHistory> results = this.mapper.selectPage(page, query);
int limit = this.config.get(HubbleOptions.EXECUTE_HISTORY_SHOW_LIMIT);
@@ -169,8 +171,9 @@ public class ExecuteHistoryService {
@Async
@Scheduled(fixedDelay = 24 * 60 * 60 * 1000)
- @Transactional(isolation = Isolation.READ_COMMITTED)
public void removeExceedLimit() {
+ // Do not wrap the complete cleanup in one transaction. Without an
+ // outer transaction, each bounded mapper DELETE commits independently.
int limit = this.config.get(HubbleOptions.EXECUTE_HISTORY_SHOW_LIMIT);
this.mapper.deleteExceedLimit(limit);
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/GremlinCollectionService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/GremlinCollectionService.java
index 66afe73ac..963c77e5e 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/GremlinCollectionService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/GremlinCollectionService.java
@@ -79,7 +79,7 @@ public class GremlinCollectionService {
// order by relativity
checkSingleOrder(nameOrderAsc, timeOrderAsc);
return this.mapper.selectByContentInPage(page, graphSpace, graph,
- content, type);
+ value, type);
}
} else {
// Select all
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/QueryService.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/QueryService.java
index 67ae3fb41..0def0ab4a 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/QueryService.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/query/QueryService.java
@@ -66,6 +66,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.function.BiConsumer;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
@@ -256,13 +257,8 @@ public class QueryService {
private String optimize(String content) {
// Optimize each gremlin statemnts
int limit = this.config.get(HubbleOptions.GREMLIN_SUFFIX_LIMIT);
- String[] originalParts = StringUtils.split(content, ";");
- String[] optimizeParts = new String[originalParts.length];
- for (int i = 0; i < originalParts.length; i++) {
- String part = originalParts[i];
- optimizeParts[i] = GremlinUtil.optimizeLimit(part, limit);
- }
- return StringUtils.join(optimizeParts, ";");
+ // The statement splitting is quote-aware, see GremlinUtil
+ return GremlinUtil.optimizeLimit(content, limit);
}
private ResultSet executeGremlin(String gremlin, HugeClient client) {
@@ -341,6 +337,20 @@ public class QueryService {
continue;
}
Object object = result.getObject();
+ if (object instanceof Map) {
+ List<Object> elements =
+ unwrapCypherElements((Map<?, ?>) object);
+ if (elements != null) {
+ for (Object element : elements) {
+ Type elementType = element instanceof Vertex ?
+ Type.VERTEX : Type.EDGE;
+ typeVotes.compute(elementType,
+ (k, v) -> v == null ? 1 : v + 1);
+ typedData.add(element);
+ }
+ continue;
+ }
+ }
Type type;
if (object instanceof Vertex) {
type = Type.VERTEX;
@@ -427,6 +437,61 @@ public class QueryService {
}
}
+ /**
+ * Cypher rows arrive as a map keyed by the RETURN variables whose values
+ * are _type-tagged node/relationship maps. Unwrap them into detached
+ * Vertex/Edge instances so such results can render as a graph; return
+ * null when the row is not purely graph elements.
+ */
+ private static List<Object> unwrapCypherElements(Map<?, ?> row) {
+ if (row.isEmpty()) {
+ return null;
+ }
+ List<Object> elements = new ArrayList<>(row.size());
+ for (Object value : row.values()) {
+ Object element = toGraphElement(value);
+ if (element == null) {
+ return null;
+ }
+ elements.add(element);
+ }
+ return elements;
+ }
+
+ private static Object toGraphElement(Object value) {
+ if (!(value instanceof Map)) {
+ return null;
+ }
+ Map<?, ?> map = (Map<?, ?>) value;
+ Object type = map.get("_type");
+ if ("node".equals(type)) {
+ Vertex vertex = new Vertex(String.valueOf(map.get("_label")));
+ copyElementProperties(map, vertex::property);
+ vertex.id(map.get("_id"));
+ return vertex;
+ }
+ if ("relationship".equals(type)) {
+ Edge edge = new Edge(String.valueOf(map.get("_label")));
+ edge.id(String.valueOf(map.get("_id")));
+ edge.sourceId(map.get("_outV"));
+ edge.targetId(map.get("_inV"));
+ copyElementProperties(map, edge::property);
+ return edge;
+ }
+ return null;
+ }
+
+ private static void copyElementProperties(Map<?, ?> map,
+ BiConsumer<String, Object> setter) {
+ for (Map.Entry<?, ?> entry : map.entrySet()) {
+ String key = String.valueOf(entry.getKey());
+ if (key.startsWith("_") || entry.getValue() == null) {
+ continue;
+ }
+ setter.accept(key, entry.getValue());
+ }
+ }
+
private GraphView buildGraphView(TypedResult result, HugeClient client) {
List<Object> data = result.getData();
if (!result.getType().isGraph() || CollectionUtils.isEmpty(data)) {
@@ -458,20 +523,28 @@ public class QueryService {
}
}
- if (!edges.isEmpty()) {
- if (vertices.isEmpty()) {
- vertices = this.verticesOfEdge(result, edges, client);
+ boolean enrichmentSucceeded = true;
+ try {
+ if (!edges.isEmpty()) {
+ if (vertices.isEmpty()) {
+ vertices = this.verticesOfEdge(result, edges, client);
+ } else {
+ // TODO: reduce the number of requests
+ vertices.putAll(this.verticesOfEdge(result, edges, client));
+ }
} else {
- // TODO: reduce the number of requests
- vertices.putAll(this.verticesOfEdge(result, edges, client));
- }
- } else {
- if (!vertices.isEmpty()) {
- edges = this.edgesOfVertex(result, vertices, client);
+ if (!vertices.isEmpty()) {
+ edges = this.edgesOfVertex(result, vertices, client);
+ }
}
+ } catch (Exception e) {
+ // The linked-element lookup is best-effort decoration; keep the
+ // elements already returned when it fails (e.g. gremlin rejected)
+ log.warn("Failed to enrich graph view with linked elements", e);
+ enrichmentSucceeded = false;
}
- if (!edges.isEmpty()) {
+ if (!edges.isEmpty() && enrichmentSucceeded) {
Ex.check(!vertices.isEmpty(),
"gremlin.edges.linked-vertex.not-exist");
}
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/GremlinUtil.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/GremlinUtil.java
index 55d21d837..c51504abf 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/GremlinUtil.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/GremlinUtil.java
@@ -18,9 +18,7 @@
package org.apache.hugegraph.util;
-import java.util.ArrayList;
import java.util.LinkedHashSet;
-import java.util.List;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -312,31 +310,181 @@ public final class GremlinUtil {
public static String optimizeLimit(String gremlin, int limit) {
// FIXME: Do not treat suffix matching as a resource-safety boundary. Decide on
// server-side timeout/result/byte limits without changing aggregation semantics.
- String[] rawLines = StringUtils.split(gremlin, "\n");
- List<String> newLines = new ArrayList<>(rawLines.length);
- for (String rawLine : rawLines) {
- boolean ignored = IGNORED_PATTERNS.stream().anyMatch(pattern -> {
- return pattern.matcher(rawLine).find();
- });
- if (ignored) {
- newLines.add(rawLine);
+ /*
+ * Scan the script with quote awareness: track single-quote,
+ * double-quote and backslash-escape state so that ';' and true
+ * statement-ending newlines are treated as separators only when
+ * outside quotes.
+ * The suffix-pattern matching and '.limit(N)' appending are
+ * applied only to true statement tails (outside quotes).
+ */
+ StringBuilder result = new StringBuilder(gremlin.length());
+ StringBuilder statement = new StringBuilder();
+ StringBuilder matchable = new StringBuilder();
+ boolean inSingleQuote = false;
+ boolean inDoubleQuote = false;
+ boolean inLineComment = false;
+ boolean inBlockComment = false;
+ boolean escaped = false;
+ for (int i = 0; i < gremlin.length(); i++) {
+ char ch = gremlin.charAt(i);
+ if (escaped) {
+ statement.append(ch);
+ matchable.append(ch);
+ escaped = false;
continue;
}
- boolean matched = false;
- for (Pattern pattern : LIMIT_PATTERNS) {
- Matcher matcher = pattern.matcher(rawLine);
- if (matcher.find()) {
- matched = true;
- break;
+ if (inLineComment) {
+ if (ch == '\n') {
+ if (chainContinues(gremlin, i)) {
+ statement.append(ch);
+ matchable.append(ch);
+ } else {
+ result.append(optimizeStatement(statement.toString(),
+ matchable.toString(),
+ limit));
+ result.append(ch);
+ statement.setLength(0);
+ matchable.setLength(0);
+ }
+ inLineComment = false;
+ } else {
+ statement.append(ch);
+ matchable.append(' ');
}
+ continue;
+ }
+ if (inBlockComment) {
+ statement.append(ch);
+ matchable.append(' ');
+ if (ch == '*' && i + 1 < gremlin.length() &&
+ gremlin.charAt(i + 1) == '/') {
+ statement.append('/');
+ matchable.append(' ');
+ i++;
+ inBlockComment = false;
+ }
+ continue;
+ }
+ if (!inSingleQuote && !inDoubleQuote &&
+ (gremlin.startsWith("'''", i) ||
+ gremlin.startsWith("\"\"\"", i))) {
+ return gremlin;
+ }
+ if (!inSingleQuote && !inDoubleQuote && ch == '$' &&
+ i + 1 < gremlin.length() &&
+ gremlin.charAt(i + 1) == '/') {
+ return gremlin;
+ }
+ if (!inSingleQuote && !inDoubleQuote && ch == '/' &&
+ i + 1 < gremlin.length()) {
+ char next = gremlin.charAt(i + 1);
+ if (next == '/' || next == '*') {
+ statement.append(ch).append(next);
+ matchable.append(" ");
+ i++;
+ inLineComment = next == '/';
+ inBlockComment = next == '*';
+ continue;
+ }
+ // This may be a slashy literal or division. Both need a real
+ // Groovy lexer to distinguish safely.
+ return gremlin;
+ }
+ if ((inSingleQuote || inDoubleQuote) && ch == '\\') {
+ statement.append(ch);
+ matchable.append(ch);
+ escaped = true;
+ continue;
+ }
+ if (ch == '\'' && !inDoubleQuote) {
+ inSingleQuote = !inSingleQuote;
+ statement.append(ch);
+ matchable.append(ch);
+ continue;
+ }
+ if (ch == '"' && !inSingleQuote) {
+ inDoubleQuote = !inDoubleQuote;
+ statement.append(ch);
+ matchable.append(ch);
+ continue;
+ }
+ boolean separator = ch == ';' ||
+ ch == '\n' &&
+ !chainContinues(gremlin, i);
+ if (separator && !inSingleQuote && !inDoubleQuote) {
+ result.append(optimizeStatement(statement.toString(),
+ matchable.toString(), limit));
+ result.append(ch);
+ statement.setLength(0);
+ matchable.setLength(0);
+ continue;
+ }
+ statement.append(ch);
+ matchable.append(ch);
+ }
+ if (inSingleQuote || inDoubleQuote || inBlockComment) {
+ // Unclosed quote: never append '.limit(N)' inside a string
+ result.append(statement);
+ } else {
+ result.append(optimizeStatement(statement.toString(),
+ matchable.toString(), limit));
+ }
+ return result.toString();
+ }
+
+ private static boolean chainContinues(String gremlin, int newline) {
+ for (int i = newline + 1; i < gremlin.length(); i++) {
+ char next = gremlin.charAt(i);
+ if (Character.isWhitespace(next)) {
+ continue;
}
- if (matched) {
- newLines.add(rawLine + ".limit(" + limit + ")");
- } else {
- newLines.add(rawLine);
+ if (next == '/' && i + 1 < gremlin.length()) {
+ char afterSlash = gremlin.charAt(i + 1);
+ if (afterSlash == '/') {
+ int lineEnd = gremlin.indexOf('\n', i + 2);
+ if (lineEnd < 0) {
+ return false;
+ }
+ i = lineEnd;
+ continue;
+ }
+ if (afterSlash == '*') {
+ int commentEnd = gremlin.indexOf("*/", i + 2);
+ if (commentEnd < 0) {
+ return false;
+ }
+ i = commentEnd + 1;
+ continue;
+ }
+ }
+ return next == '.';
+ }
+ return false;
+ }
+
+ private static String optimizeStatement(String statement,
+ String matchable, int limit) {
+ int codeEnd = matchable.length();
+ while (codeEnd > 0 && Character.isWhitespace(
+ matchable.charAt(codeEnd - 1))) {
+ codeEnd--;
+ }
+ String code = matchable.substring(0, codeEnd);
+ boolean ignored = IGNORED_PATTERNS.stream().anyMatch(pattern -> {
+ return pattern.matcher(code).find();
+ });
+ if (ignored) {
+ return statement;
+ }
+ for (Pattern pattern : LIMIT_PATTERNS) {
+ Matcher matcher = pattern.matcher(code);
+ if (matcher.find()) {
+ return statement.substring(0, codeEnd) + ".limit(" + limit +
+ ")" + statement.substring(codeEnd);
}
}
- return StringUtils.join(newLines, "\n");
+ return statement;
}
private static Set<Pattern> compile(Set<String> texts) {
diff --git a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/PageUtil.java b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/PageUtil.java
index 523410fe6..4dbb449ef 100644
--- a/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/PageUtil.java
+++ b/hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/util/PageUtil.java
@@ -29,6 +29,30 @@ public final class PageUtil {
public static final int MAX_PAGE_SIZE = 500;
+ /*
+ * Hard cap used instead of an unbounded query when the frontend
+ * passes page_size=-1 (meaning "all records")
+ */
+ public static final int HARD_CAP = 10000;
+
+ /**
+ * Bound the requested page size: page_size=-1 is accepted from the
+ * frontend but is turned into a hard cap instead of unbounded.
+ */
+ public static int boundedSize(int requested) {
+ if (requested < -1) {
+ throw new IllegalArgumentException("page_size must be >= -1");
+ }
+ return requested == -1 ? HARD_CAP : requested;
+ }
+
+ public static long boundedSize(long requested) {
+ if (requested < -1L) {
+ throw new IllegalArgumentException("page_size must be >= -1");
+ }
+ return requested == -1L ? HARD_CAP : requested;
+ }
+
public static void checkPage(int pageNo, int pageSize) {
boolean invalidSize = pageSize != -1 &&
(pageSize < 1 || pageSize > MAX_PAGE_SIZE);
diff --git a/hugegraph-hubble/hubble-be/src/main/resources/database/schema.sql b/hugegraph-hubble/hubble-be/src/main/resources/database/schema.sql
index fbfbac687..df804edd0 100644
--- a/hugegraph-hubble/hubble-be/src/main/resources/database/schema.sql
+++ b/hugegraph-hubble/hubble-be/src/main/resources/database/schema.sql
@@ -89,6 +89,7 @@ CREATE TABLE IF NOT EXISTS `execute_history` (
);
CREATE INDEX IF NOT EXISTS `execute_history_conn_id` ON `execute_history`(`conn_id`);
+CREATE INDEX IF NOT EXISTS `execute_history_graph_create_time` ON `execute_history`(`graphspace`, `graph`, `create_time`);
// DROP TABLE IF EXISTS `edit_history`;
@@ -120,7 +121,7 @@ CREATE TABLE IF NOT EXISTS `gremlin_collection` (
`content` VARCHAR(65535) NOT NULL,
`create_time` DATETIME(6) NOT NULL,
PRIMARY KEY (`id`),
- UNIQUE (`conn_id`, `name`)
+ UNIQUE (`graphspace`, `graph`, `name`, `type`)
);
CREATE INDEX IF NOT EXISTS `gremlin_collection_conn_id` ON `gremlin_collection`(`conn_id`);
@@ -145,6 +146,7 @@ CREATE TABLE IF NOT EXISTS `file_mapping` (
UNIQUE (`conn_id`, `job_id`, `name`)
);
CREATE INDEX IF NOT EXISTS `file_mapping_conn_id` ON `file_mapping`(`conn_id`);
+CREATE INDEX IF NOT EXISTS `file_mapping_job_id` ON `file_mapping`(`job_id`);
CREATE TABLE IF NOT EXISTS `load_task` (
`id` INT NOT NULL AUTO_INCREMENT,
@@ -201,6 +203,9 @@ CREATE TABLE IF NOT EXISTS `async_task` (
CREATE INDEX IF NOT EXISTS `load_task_conn_id` ON `load_task`(`conn_id`);
CREATE INDEX IF NOT EXISTS `load_task_file_id` ON `load_task`(`file_id`);
+CREATE INDEX IF NOT EXISTS `load_task_job_id` ON `load_task`(`job_id`);
+CREATE INDEX IF NOT EXISTS `job_manager_graph_create_time` ON `job_manager`(`graphspace`, `graph`, `create_time`);
+CREATE INDEX IF NOT EXISTS `async_task_graph` ON `async_task`(`graphspace`, `graph`);
CREATE TABLE IF NOT EXISTS `datasource` (
`id` INT NOT NULL AUTO_INCREMENT,
diff --git a/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages.properties b/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages.properties
index 3986e567e..e4b96eb0c 100644
--- a/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages.properties
+++ b/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages.properties
@@ -181,6 +181,23 @@ load.file-mapping.load-parameter.max-insert-error.illegal=The load parameter 'ma
load.file-mapping.load-parameter.insert-timeout.illegal=The load parameter 'insert-timeout' is illegal
load.file-mapping.load-parameter.retry-times.illegal=The load parameter 'retry-times' is illegal
load.file-mapping.load-parameter.retry-interval.illegal=The load parameter 'retry-interval' is illegal
+load.upload.file.no-permission=Cannot upload files because the current job status does not allow it
+load.upload.merge-file.failed=Failed to merge the uploaded file parts
+load.upload.move-file.failed=Failed to move the uploaded file
+load.upload.rename-file.failed=Failed to rename the uploaded file
+load.upload.delete-temp-dir.failed=Failed to delete the temporary upload directory
+deleted.file.no-permission=Cannot delete the file because the current job status does not allow it
+file-mapping.not-exist.id=The file does not exist with id {0}
+load.file-mapping.delete.task-in-running=Cannot delete the file with id {0} because an associated load task is still running, please stop the task first
+load.task.not-exist.id=The load task does not exist with id {0}
+load.task.reached-limit=The number of load tasks has reached the limit {0}
+load.task.create.no-permission=Cannot create the load task because the current job status does not allow it
+load.task.start.no-permission=Cannot start the load task because the current job status does not allow it
+load.task.pause.no-permission=Cannot pause the load task because the current job status does not allow it
+load.task.resume.no-permission=Cannot resume the load task because the current job status does not allow it
+load.task.stop.no-permission=Cannot stop the load task because the current job status does not allow it
+load.task.retry.no-permission=Cannot retry the load task because the current job status does not allow it
+load.task.delete.task-in-running=Cannot delete the load task with id {0} because it is still running, please stop it first
load.build-task.failed=Build load task failed
# job manager
@@ -190,6 +207,12 @@ job.manager.job-name.unmatch-regex=Invalid job name, valid name is up to 48 alph
job.manager.job-remarks.reached-limit=Can't add job remarks because has length limit
job.manager.job-remarks.unmatch-regex=Invalid job name, valid name is up to 200 alpha-numeric characters and underscores
job.manager.status.unexpected=Job status doesn't meet expectations, it should be {0}, actually {1}
+job.manager.not-exist.id=The job does not exist with id {0}
+job-manager.not-exist.id=The job does not exist with id {0}
+job.manager.reached-limit=The number of jobs has reached the limit {0}
+
+# Async task
+async.task.not-exist.id=The async task does not exist with id {0}
# Internal Exceptions
common.unknown.enum.type=Unknown type {0} for enum {1}
@@ -228,4 +251,4 @@ service.url.parse.error=Parse host info ({0}) error?please check the url in \
service config
service.manual.disable.modify=Disallown modify manual service
-auth.login.throttled=Too many login failures. Please retry in {0} seconds
+auth.login.throttled=Too many login failures. Please retry in {0}s
diff --git a/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages_zh_CN.properties b/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages_zh_CN.properties
index e412e5f1f..beb88bbe0 100644
--- a/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages_zh_CN.properties
+++ b/hugegraph-hubble/hubble-be/src/main/resources/i18n/messages_zh_CN.properties
@@ -181,6 +181,23 @@ load.file-mapping.load-parameter.max-insert-error.illegal=导入参数 允许最
load.file-mapping.load-parameter.insert-timeout.illegal=导入参数 插入超时时间/s 不合法
load.file-mapping.load-parameter.retry-times.illegal=导入参数 插入失败重试次数 不合法
load.file-mapping.load-parameter.retry-interval.illegal=导入参数 插入失败重试间隔/s 不合法
+load.upload.file.no-permission=当前导入任务的状态不允许上传文件
+load.upload.merge-file.failed=合并上传的文件分片失败
+load.upload.move-file.failed=移动上传的文件失败
+load.upload.rename-file.failed=重命名上传的文件失败
+load.upload.delete-temp-dir.failed=删除上传的临时目录失败
+deleted.file.no-permission=当前导入任务的状态不允许删除文件
+file-mapping.not-exist.id=不存在 id 为 {0} 的文件
+load.file-mapping.delete.task-in-running=不能删除 id 为 {0} 的文件，其关联的导入任务正在运行，请先停止该任务
+load.task.not-exist.id=不存在 id 为 {0} 的导入任务
+load.task.reached-limit=导入任务的数量已达到上限 {0}
+load.task.create.no-permission=当前导入任务的状态不允许创建导入任务
+load.task.start.no-permission=当前导入任务的状态不允许开始导入
+load.task.pause.no-permission=当前导入任务的状态不允许暂停导入
+load.task.resume.no-permission=当前导入任务的状态不允许恢复导入
+load.task.stop.no-permission=当前导入任务的状态不允许停止导入
+load.task.retry.no-permission=当前导入任务的状态不允许重试导入
+load.task.delete.task-in-running=不能删除 id 为 {0} 的导入任务，它正在运行，请先停止该任务
load.build-task.failed=构建导入任务失败
# job manager
@@ -190,6 +207,12 @@ job.manager.job-name.unmatch-regex=任务名不合法，任务名允许字母、
job.manager.job-remarks.reached-limit=任务描述长度限制48个字符
job.manager.job-remarks.unmatch-regex=任务描述不合法，任务描述允许字母、数字、中文、下划线，最多 200 个字符
job.manager.status.unexpected=当前的任务状态不符合预期，应该是 {0}，实际是 {1}
+job.manager.not-exist.id=不存在 id 为 {0} 的导入任务
+job-manager.not-exist.id=不存在 id 为 {0} 的导入任务
+job.manager.reached-limit=导入任务的数量已达到上限 {0}
+
+# Async task
+async.task.not-exist.id=不存在 id 为 {0} 的异步任务
# Internal Exceptions
common.unknown.enum.type=枚举 {1} 的值 {0} 未知
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/config/LegacySchemaStartupTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/config/LegacySchemaStartupTest.java
new file mode 100644
index 000000000..194286ff4
--- /dev/null
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/config/LegacySchemaStartupTest.java
@@ -0,0 +1,233 @@
+/*
+ *
+ * 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 org.apache.hugegraph.config;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import javax.sql.DataSource;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Import;
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.jdbc.datasource.init.ScriptUtils;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import org.apache.hugegraph.driver.HugeClient;
+import org.apache.hugegraph.entity.enums.ExecuteType;
+import org.apache.hugegraph.entity.query.ExecuteHistory;
+import org.apache.hugegraph.entity.query.GremlinCollection;
+import org.apache.hugegraph.options.HubbleOptions;
+import org.apache.hugegraph.service.query.ExecuteHistoryService;
+import org.apache.hugegraph.service.query.GremlinCollectionService;
+import org.apache.hugegraph.testutil.Assert;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = LegacySchemaStartupTest.TestConfiguration.class,
+ webEnvironment = SpringBootTest.WebEnvironment.NONE,
+ properties = {
+ "spring.datasource.url=" + LegacySchemaStartupTest.JDBC_URL,
+ "spring.datasource.driver-class-name=org.h2.Driver",
+ "spring.datasource.username=sa",
+ "spring.datasource.password=",
+ "spring.datasource.schema=" +
+ "file:src/main/resources/database/schema.sql",
+ "spring.datasource.initialization-mode=always",
+ "spring.datasource.hikari.maximum-pool-size=2",
+ "spring.autoconfigure.exclude=" +
+ "org.mybatis.spring.boot.autoconfigure.MybatisAutoConfiguration",
+ "mybatis.configuration.map-underscore-to-camel-case=true",
+ "mybatis.configuration.use-generated-keys=true",
+ "mybatis-plus.type-enums-package=" +
+ "org.apache.hugegraph.entity.enums"
+ })
+public class LegacySchemaStartupTest {
+
+ static final String JDBC_URL =
+ "jdbc:h2:./target/legacy-schema-startup;DB_CLOSE_DELAY=-1";
+
+ static {
+ prepareLegacyDatabase();
+ }
+
+ @Autowired
+ private DataSource dataSource;
+ @Autowired
+ private DatabaseSchemaMigrator migrator;
+ @Autowired
+ private ExecuteHistoryService historyService;
+ @Autowired
+ private GremlinCollectionService collectionService;
+ @MockBean
+ private HugeConfig config;
+
+ private HugeClient client;
+
+ @Before
+ public void setup() throws Exception {
+ Mockito.when(this.config.get(HubbleOptions.EXECUTE_HISTORY_SHOW_LIMIT))
+ .thenReturn(500);
+ this.client = Mockito.mock(HugeClient.class);
+ Mockito.when(this.client.getGraphSpaceName()).thenReturn("DEFAULT");
+ Mockito.when(this.client.getGraphName()).thenReturn("legacygraph");
+ try (Connection conn = this.dataSource.getConnection()) {
+ this.migrator.migrate(conn);
+ }
+ }
+
+ @Test
+ public void testLegacyDatabaseStartsAndRowsAreVisibleThroughServices()
+ throws Exception {
+ IPage<ExecuteHistory> histories = this.historyService.list(
+ this.client, ExecuteType.GREMLIN.getValue(), 1L, 10L, false);
+ Assert.assertEquals(1L, histories.getTotal());
+ Assert.assertEquals("g.V()", histories.getRecords().get(0).getContent());
+
+ IPage<GremlinCollection> collections = this.collectionService.list(
+ this.client, null, "GREMLIN", null, false, 1L, 10L);
+ Assert.assertEquals(1L, collections.getTotal());
+ Assert.assertEquals("saved", collections.getRecords().get(0).getName());
+
+ try (Connection conn = this.dataSource.getConnection()) {
+ this.assertScopedRows(conn);
+ }
+ }
+
+ @Test
+ public void testSavedQueryConstraintMatchesApplicationScope()
+ throws Exception {
+ try (Connection conn = this.dataSource.getConnection();
+ Statement statement = conn.createStatement()) {
+ statement.execute("DELETE FROM `gremlin_collection` " +
+ "WHERE `graph` = 'othergraph'");
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`conn_id`, `graphspace`, `graph`, `name`, " +
+ "`type`, `content`, `create_time`) VALUES " +
+ "(1, 'DEFAULT', 'othergraph', 'saved', " +
+ "'GREMLIN', 'g.E()', CURRENT_TIMESTAMP)");
+ try (ResultSet rows = statement.executeQuery(
+ "SELECT COUNT(*) FROM `gremlin_collection` WHERE " +
+ "`graphspace` = 'DEFAULT' AND `graph` = 'othergraph' " +
+ "AND `name` = 'saved' AND `type` = 'GREMLIN'")) {
+ Assert.assertTrue(rows.next());
+ Assert.assertEquals(1L, rows.getLong(1));
+ }
+ Assert.assertThrows(SQLException.class, () ->
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`conn_id`, `graphspace`, `graph`, " +
+ "`name`, `type`, `content`, " +
+ "`create_time`) VALUES " +
+ "(1, 'DEFAULT', 'legacygraph', " +
+ "'saved', 'GREMLIN', 'g.E()', " +
+ "CURRENT_TIMESTAMP)"));
+ }
+ }
+
+ @Test
+ public void testLegacyScopeBackfillRequiresAllTargetColumns()
+ throws Exception {
+ try (Connection conn = this.dataSource.getConnection();
+ Statement statement = conn.createStatement()) {
+ statement.execute("DROP TABLE IF EXISTS `scope_guard`");
+ statement.execute("CREATE TABLE `scope_guard` (`conn_id` INT)");
+ Assert.assertFalse(this.migrator.scopeColumnsExist(conn,
+ "scope_guard"));
+ statement.execute("ALTER TABLE `scope_guard` ADD COLUMN " +
+ "`graphspace` VARCHAR(48)");
+ statement.execute("ALTER TABLE `scope_guard` ADD COLUMN " +
+ "`graph` VARCHAR(48)");
+ Assert.assertTrue(this.migrator.scopeColumnsExist(conn,
+ "scope_guard"));
+ statement.execute("DROP TABLE `scope_guard`");
+ }
+ }
+
+ @Test
+ public void testMysqlUsesDropIndexForLegacyUniqueKeys() {
+ Assert.assertEquals("ALTER TABLE `gremlin_collection` DROP INDEX " +
+ "`legacy_name_key`",
+ DatabaseSchemaMigrator.dropUniqueKeySql(
+ "MySQL", "legacy_name_key"));
+ }
+
+ private void assertScopedRows(Connection conn) throws Exception {
+ try (Statement statement = conn.createStatement();
+ ResultSet rows = statement.executeQuery(
+ "SELECT `conn_id`, `graphspace`, `graph` FROM " +
+ "`execute_history`")) {
+ Assert.assertTrue(rows.next());
+ Assert.assertNull(rows.getObject(1));
+ Assert.assertEquals("DEFAULT", rows.getString(2));
+ Assert.assertEquals("legacygraph", rows.getString(3));
+ }
+ }
+
+ private static void prepareLegacyDatabase() {
+ Path database = Paths.get("target/legacy-schema-startup.mv.db");
+ try {
+ Files.createDirectories(database.getParent());
+ Files.deleteIfExists(database);
+ try (Connection conn = DriverManager.getConnection(JDBC_URL,
+ "sa", "");
+ Statement statement = conn.createStatement()) {
+ ScriptUtils.executeSqlScript(conn, new FileSystemResource(
+ Paths.get("src/test/resources/database/schema.sql")));
+ statement.execute("INSERT INTO `graph_connection` " +
+ "(`name`, `graph`, `host`, `port`, " +
+ "`create_time`) VALUES ('legacy', " +
+ "'legacygraph', 'localhost', 8080, " +
+ "CURRENT_TIMESTAMP)");
+ statement.execute("INSERT INTO `execute_history` " +
+ "(`execute_type`, `content`, " +
+ "`execute_status`, `duration`, " +
+ "`create_time`) VALUES " +
+ "(0, 'g.V()', 1, 10, CURRENT_TIMESTAMP)");
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`name`, `content`, `create_time`) VALUES " +
+ "('saved', 'g.E()', CURRENT_TIMESTAMP)");
+ }
+ } catch (Exception e) {
+ throw new ExceptionInInitializerError(e);
+ }
+ }
+
+ @SpringBootConfiguration
+ @EnableAutoConfiguration
+ @MapperScan("org.apache.hugegraph.mapper.query")
+ @Import({DatabaseSchemaMigrator.class, MybatisPlusConfig.class,
+ ExecuteHistoryService.class, GremlinCollectionService.class})
+ public static class TestConfiguration {
+ }
+}
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/controller/op/HStoreControllerTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/controller/op/HStoreControllerTest.java
new file mode 100644
index 000000000..92b1462be
--- /dev/null
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/controller/op/HStoreControllerTest.java
@@ -0,0 +1,114 @@
+/*
+ *
+ * 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 org.apache.hugegraph.controller.op;
+
+import java.util.Map;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import org.apache.hugegraph.driver.HStoreManager;
+import org.apache.hugegraph.driver.HugeClient;
+import org.apache.hugegraph.exception.ServerCapabilityUnavailableException;
+import org.apache.hugegraph.exception.ServerException;
+import org.apache.hugegraph.testutil.Assert;
+
+public class HStoreControllerTest {
+
+ @Test
+ public void testMissingApiStatusesMapToCapabilityUnavailable() {
+ for (int status : new int[]{404, 405, 501}) {
+ ServerException server = serverException(status);
+ RuntimeException mapped = HStoreController.mapHstoreFailure(
+ server, "node-2", "node-1");
+ Assert.assertTrue(mapped instanceof
+ ServerCapabilityUnavailableException);
+ Assert.assertSame(server, mapped.getCause());
+ }
+ }
+
+ @Test
+ public void testOtherServerFailuresRemainUnchanged() {
+ for (int status : new int[]{400, 401, 403, 409, 500, 503}) {
+ ServerException server = serverException(status);
+ Assert.assertSame(server,
+ HStoreController.mapHstoreFailure(server));
+ }
+ }
+
+ @Test
+ public void testStartupFailureReportsSuccessfulAndFailedNodes() {
+ this.assertPartialNodeFailure(true);
+ }
+
+ @Test
+ public void testShutdownFailureReportsSuccessfulAndFailedNodes() {
+ this.assertPartialNodeFailure(false);
+ }
+
+ private void assertPartialNodeFailure(boolean startup) {
+ HugeClient client = Mockito.mock(HugeClient.class);
+ HStoreManager manager = Mockito.mock(HStoreManager.class);
+ Mockito.when(client.hStoreManager()).thenReturn(manager);
+ ServerException server = serverException(404);
+ if (startup) {
+ Mockito.doThrow(server).when(manager).nodeStartup("node-2");
+ } else {
+ Mockito.doThrow(server).when(manager).nodeShutdown("node-2");
+ }
+ HStoreController controller = new TestHStoreController(client);
+ Map<String, java.util.List<String>> request = ImmutableMap.of(
+ "nodes", ImmutableList.of("node-1", "node-2"));
+
+ try {
+ if (startup) {
+ controller.nodesStartup(request);
+ } else {
+ controller.nodesShutdown(request);
+ }
+ Assert.fail("Expected unavailable HStore capability");
+ } catch (ServerCapabilityUnavailableException e) {
+ Assert.assertSame(server, e.getCause());
+ Assert.assertEquals("node-2", e.args()[0]);
+ Assert.assertEquals(ImmutableList.of("node-1"), e.args()[1]);
+ }
+ }
+
+ private static ServerException serverException(int status) {
+ ServerException server = new ServerException("upstream-" + status);
+ server.status(status);
+ return server;
+ }
+
+ private static final class TestHStoreController extends HStoreController {
+
+ private final HugeClient client;
+
+ private TestHStoreController(HugeClient client) {
+ this.client = client;
+ }
+
+ @Override
+ protected HugeClient authClient(String graphSpace, String graph) {
+ return this.client;
+ }
+ }
+}
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/load/IngestTransactionIntegrationTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/load/IngestTransactionIntegrationTest.java
index 1a57d89a4..3d90b6f01 100644
--- a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/load/IngestTransactionIntegrationTest.java
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/load/IngestTransactionIntegrationTest.java
@@ -20,6 +20,11 @@ package org.apache.hugegraph.service.load;
import java.util.Collections;
import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.sql.DataSource;
@@ -55,6 +60,7 @@ import org.apache.hugegraph.entity.load.LoadParameter;
import org.apache.hugegraph.entity.load.LoadTask;
import org.apache.hugegraph.handler.LoadTaskExecutor;
import org.apache.hugegraph.loader.executor.LoadOptions;
+import org.apache.hugegraph.options.HubbleOptions;
import org.apache.hugegraph.service.schema.EdgeLabelService;
import org.apache.hugegraph.service.schema.VertexLabelService;
import org.apache.hugegraph.testutil.Assert;
@@ -79,14 +85,21 @@ import org.apache.hugegraph.testutil.Assert;
})
public class IngestTransactionIntegrationTest {
+ private static final String UPLOAD_ROOT =
+ "/tmp/hubble-ingest-transaction-test";
+
@Autowired
private JobManagerService jobService;
@Autowired
+ private FileMappingService mappingService;
+ @Autowired
private TestLoadTaskService taskService;
@Autowired
private RecordingLoadTaskExecutor executor;
@Autowired
private DataSource dataSource;
+ @Autowired
+ private HugeConfig config;
@MockBean
private VertexLabelService vertexLabelService;
@MockBean
@@ -100,6 +113,8 @@ public class IngestTransactionIntegrationTest {
this.cleanup();
this.executor.reset();
this.taskService.failTaskInsert(false);
+ Mockito.when(this.config.get(HubbleOptions.UPLOAD_FILE_LOCATION))
+ .thenReturn(UPLOAD_ROOT);
}
@After
@@ -180,6 +195,99 @@ public class IngestTransactionIntegrationTest {
Assert.assertEquals("canary-server-token", task.getOptions().token);
}
+ @Test
+ public void testConcurrentQuotaChangesAreAtomic() throws Exception {
+ JobManager job = this.job();
+ job.setJobSize(0L);
+ this.jobService.save(job);
+ FileMapping first = this.persistedMapping(job.getId(), "first.csv");
+ FileMapping second = this.persistedMapping(job.getId(), "second.csv");
+
+ this.runConcurrently(
+ () -> this.jobService.completeUpload(first),
+ () -> this.jobService.completeUpload(second));
+
+ Assert.assertEquals(16L, this.jobService.get(job.getId()).getJobSize());
+
+ this.runConcurrently(
+ () -> this.jobService.deleteMappings(
+ job.getId(), Collections.singletonList(first)),
+ () -> this.jobService.deleteMappings(
+ job.getId(), Collections.singletonList(second)));
+
+ Assert.assertEquals(0L, this.jobService.get(job.getId()).getJobSize());
+ Assert.assertEquals(0, this.count("file_mapping"));
+ }
+
+ @Test
+ public void testQuotaFailureRollsBackMappingDeletion() {
+ FileMapping mapping = this.persistedMapping(999, "orphan.csv");
+
+ Assert.assertThrows(RuntimeException.class, () ->
+ this.jobService.deleteMappings(
+ 999, Collections.singletonList(mapping)));
+
+ Assert.assertEquals(1, this.count("file_mapping"));
+ }
+
+ @Test
+ public void testQuotaFailureRollsBackUploadCompletion() {
+ FileMapping mapping = this.mapping();
+ mapping.setJobId(999);
+ mapping.setFileStatus(FileMappingStatus.UPLOADING);
+ this.mappingService.save(mapping);
+
+ mapping.setFileStatus(FileMappingStatus.COMPLETED);
+ Assert.assertThrows(RuntimeException.class, () ->
+ this.jobService.completeUpload(mapping));
+
+ Integer status = this.jdbc.queryForObject(
+ "SELECT file_status FROM file_mapping WHERE id = ?",
+ Integer.class, mapping.getId());
+ Assert.assertEquals((int) FileMappingStatus.UPLOADING.getValue(),
+ status.intValue());
+ }
+
+ @Test
+ public void testFailedMappingDeletionDoesNotReleaseCompletedQuota() {
+ JobManager job = this.job();
+ job.setJobSize(8L);
+ this.jobService.save(job);
+ this.persistedMapping(job.getId(), "completed.csv");
+ FileMapping failed = this.persistedMapping(job.getId(), "failed.csv");
+ failed.setFileStatus(FileMappingStatus.FAILURE);
+ this.mappingService.update(failed);
+
+ this.jobService.deleteMappings(
+ job.getId(), Collections.singletonList(failed));
+
+ Assert.assertEquals(8L, this.jobService.get(job.getId()).getJobSize());
+ }
+
+ @Test
+ public void testFailedDiskCleanupLeavesRetryableOrphanRecord() {
+ JobManager job = this.job();
+ job.setJobSize(8L);
+ this.jobService.save(job);
+ FileMapping mapping = this.persistedMapping(job.getId(), "orphan.csv");
+ mapping.setPath("/outside-upload-root/orphan.csv");
+ this.mappingService.update(mapping);
+
+ this.jobService.deleteMappings(
+ job.getId(), Collections.singletonList(mapping));
+
+ Assert.assertEquals(1, this.count("file_mapping"));
+ Assert.assertEquals(-mapping.getId(),
+ this.jdbc.queryForObject(
+ "SELECT job_id FROM file_mapping WHERE id = ?",
+ Integer.class, mapping.getId()).intValue());
+
+ this.jdbc.update("UPDATE file_mapping SET path = ? WHERE id = ?",
+ UPLOAD_ROOT + "/retry/orphan.csv", mapping.getId());
+ this.mappingService.deleteOrphanedJobFiles();
+ Assert.assertEquals(0, this.count("file_mapping"));
+ }
+
@Test
public void testRejectedDispatchCommitsFailedTaskAndJobWithoutThrowing() {
this.executor.reject(true);
@@ -234,6 +342,48 @@ public class IngestTransactionIntegrationTest {
return mapping;
}
+ private FileMapping persistedMapping(int jobId, String name) {
+ FileMapping mapping = this.mapping();
+ mapping.setJobId(jobId);
+ mapping.setName(name);
+ mapping.setPath(UPLOAD_ROOT + "/" + name + "/" + name);
+ this.mappingService.save(mapping);
+ return mapping;
+ }
+
+ private void runConcurrently(Runnable first, Runnable second)
+ throws Exception {
+ ExecutorService pool = Executors.newFixedThreadPool(2);
+ CountDownLatch ready = new CountDownLatch(2);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ Future<?> firstFuture = pool.submit(
+ () -> runAfterSignal(first, ready, start));
+ Future<?> secondFuture = pool.submit(
+ () -> runAfterSignal(second, ready, start));
+ Assert.assertTrue(ready.await(5L, TimeUnit.SECONDS));
+ start.countDown();
+ firstFuture.get(10L, TimeUnit.SECONDS);
+ secondFuture.get(10L, TimeUnit.SECONDS);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ private static void runAfterSignal(Runnable action, CountDownLatch ready,
+ CountDownLatch start) {
+ ready.countDown();
+ try {
+ if (!start.await(5L, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Concurrent test start timed out");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Concurrent test interrupted", e);
+ }
+ action.run();
+ }
+
private GraphConnection connection() {
GraphConnection connection = new GraphConnection();
connection.setGraphSpace("DEFAULT");
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/query/ExecuteHistoryRetentionTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/query/ExecuteHistoryRetentionTest.java
new file mode 100644
index 000000000..2cfb7bff2
--- /dev/null
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/service/query/ExecuteHistoryRetentionTest.java
@@ -0,0 +1,188 @@
+/*
+ *
+ * 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 org.apache.hugegraph.service.query;
+
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import javax.sql.DataSource;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+
+import org.apache.hugegraph.entity.query.ExecuteHistory;
+import org.apache.hugegraph.mapper.query.ExecuteHistoryMapper;
+import org.apache.hugegraph.testutil.Assert;
+import org.mockito.Mockito;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = ExecuteHistoryRetentionTest.TestConfiguration.class,
+ webEnvironment = SpringBootTest.WebEnvironment.NONE,
+ properties = {
+ "spring.datasource.url=jdbc:h2:mem:execute-history-retention;" +
+ "DB_CLOSE_DELAY=-1",
+ "spring.datasource.driver-class-name=org.h2.Driver",
+ "spring.datasource.username=sa",
+ "spring.datasource.password=",
+ "spring.datasource.initialization-mode=never",
+ "spring.datasource.hikari.maximum-pool-size=2",
+ "spring.autoconfigure.exclude=" +
+ "org.mybatis.spring.boot.autoconfigure.MybatisAutoConfiguration",
+ "mybatis.configuration.map-underscore-to-camel-case=true",
+ "mybatis.configuration.use-generated-keys=true",
+ "mybatis-plus.type-enums-package=org.apache.hugegraph.entity.enums"
+ })
+public class ExecuteHistoryRetentionTest {
+
+ private static final String SPACE = "DEFAULT";
+
+ @Autowired
+ private ExecuteHistoryMapper mapper;
+ @Autowired
+ private DataSource dataSource;
+
+ private JdbcTemplate jdbc;
+
+ @Before
+ public void setup() {
+ this.jdbc = new JdbcTemplate(this.dataSource);
+ this.jdbc.execute("CREATE TABLE IF NOT EXISTS `execute_history` (" +
+ "`id` INT NOT NULL AUTO_INCREMENT, " +
+ "`conn_id` INT, " +
+ "`graphspace` VARCHAR(48) NOT NULL, " +
+ "`graph` VARCHAR(48) NOT NULL, " +
+ "`async_id` BIGINT NOT NULL, " +
+ "`execute_type` TINYINT NOT NULL, " +
+ "`content` VARCHAR(65535) NOT NULL, " +
+ "`text` VARCHAR(65535) NOT NULL, " +
+ "`execute_status` TINYINT NOT NULL, " +
+ "`failure_reason` VARCHAR(64) DEFAULT NULL, " +
+ "`async_status` TINYINT NOT NULL DEFAULT 0, " +
+ "`duration` BIGINT NOT NULL, " +
+ "`create_time` DATETIME(6) NOT NULL, " +
+ "PRIMARY KEY (`id`))");
+ this.jdbc.update("DELETE FROM `execute_history`");
+ }
+
+ @Test
+ public void testKeepsNewestRowsPerGraphAndDeletesTheRest() {
+ // A busy graph and a quiet one, interleaved in time
+ for (int i = 0; i < 6; i++) {
+ this.insert("busy", "busy-" + i, 1000L + (i * 1000L));
+ }
+ this.insert("quiet", "quiet-0", 1000L);
+ this.insert("quiet", "quiet-1", 2000L);
+
+ this.mapper.deleteExceedLimit(2);
+
+ // The quiet graph is untouched, the busy one keeps only the newest 2
+ Assert.assertEquals(2, this.count("quiet"));
+ Assert.assertEquals(2, this.count("busy"));
+ List<String> kept = this.jdbc.queryForList(
+ "SELECT `content` FROM `execute_history` " +
+ "WHERE `graph` = 'busy' ORDER BY `create_time` DESC",
+ String.class);
+ Assert.assertEquals(2, kept.size());
+ Assert.assertEquals("busy-5", kept.get(0));
+ Assert.assertEquals("busy-4", kept.get(1));
+ }
+
+ @Test
+ public void testKeepsEverythingWhenUnderTheLimit() {
+ this.insert("small", "only-0", 1000L);
+ this.insert("small", "only-1", 2000L);
+
+ this.mapper.deleteExceedLimit(500);
+
+ Assert.assertEquals(2, this.count("small"));
+ }
+
+ @Test
+ public void testDeletesExcessRowsInBoundedBatches() {
+ for (int i = 0; i < 505; i++) {
+ this.insert("busy", "busy-" + i, 1000L + i);
+ }
+
+ this.mapper.deleteExceedLimit(2);
+
+ Assert.assertEquals(2, this.count("busy"));
+ List<String> kept = this.jdbc.queryForList(
+ "SELECT `content` FROM `execute_history` " +
+ "WHERE `graph` = 'busy' ORDER BY `create_time` DESC",
+ String.class);
+ Assert.assertEquals("busy-504", kept.get(0));
+ Assert.assertEquals("busy-503", kept.get(1));
+ }
+
+ @Test
+ public void testRechecksExcessAfterSelectingCandidates() {
+ ExecuteHistoryMapper concurrent = Mockito.mock(
+ ExecuteHistoryMapper.class, Mockito.CALLS_REAL_METHODS);
+ ExecuteHistory scope = new ExecuteHistory();
+ scope.setGraphspace(SPACE);
+ scope.setGraph("busy");
+ Mockito.when(concurrent.findScopes()).thenReturn(
+ Collections.singletonList(scope));
+ Mockito.when(concurrent.countScope(SPACE, "busy")).thenReturn(4, 2);
+ Mockito.when(concurrent.findIdsOldestFirst(SPACE, "busy", 2))
+ .thenReturn(Arrays.asList(1L, 2L));
+
+ concurrent.deleteExceedLimit(2);
+
+ Mockito.verify(concurrent, Mockito.times(2))
+ .countScope(SPACE, "busy");
+ Mockito.verify(concurrent, Mockito.never())
+ .deleteBatchIds(Mockito.anyList());
+ }
+
+ private void insert(String graph, String content, long millis) {
+ this.jdbc.update("INSERT INTO `execute_history` " +
+ "(`graphspace`, `graph`, `async_id`, " +
+ "`execute_type`, `content`, `text`, " +
+ "`execute_status`, `async_status`, `duration`, " +
+ "`create_time`) VALUES (?, ?, 0, 1, ?, '', 1, 0, " +
+ "0, ?)",
+ SPACE, graph, content, new Timestamp(millis));
+ }
+
+ private int count(String graph) {
+ return this.jdbc.queryForObject(
+ "SELECT COUNT(*) FROM `execute_history` " +
+ "WHERE `graphspace` = ? AND `graph` = ?",
+ Integer.class, SPACE, graph);
+ }
+
+ @SpringBootConfiguration
+ @EnableAutoConfiguration
+ @EnableTransactionManagement
+ @MapperScan("org.apache.hugegraph.mapper.query")
+ public static class TestConfiguration {
+ }
+}
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingDeletionTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingDeletionTest.java
new file mode 100644
index 000000000..1656cf852
--- /dev/null
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingDeletionTest.java
@@ -0,0 +1,110 @@
+/*
+ *
+ * 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 org.apache.hugegraph.unit;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import org.apache.hugegraph.controller.load.FileMappingController;
+import org.apache.hugegraph.entity.enums.LoadStatus;
+import org.apache.hugegraph.entity.load.FileMapping;
+import org.apache.hugegraph.entity.load.LoadTask;
+import org.apache.hugegraph.exception.ExternalException;
+import org.apache.hugegraph.service.load.FileMappingService;
+import org.apache.hugegraph.service.load.JobManagerService;
+import org.apache.hugegraph.service.load.LoadTaskService;
+import org.apache.hugegraph.testutil.Assert;
+
+public class FileMappingDeletionTest {
+
+ @Test
+ public void testDeleteProtectsEveryRecoverableTaskStatus()
+ throws Exception {
+ for (LoadStatus status : new LoadStatus[]{LoadStatus.RUNNING,
+ LoadStatus.PAUSED,
+ LoadStatus.FAILED,
+ LoadStatus.STOPPED}) {
+ Fixture fixture = this.fixture(status);
+
+ Assert.assertThrows(ExternalException.class, () ->
+ fixture.controller.delete("DEFAULT", "hugegraph", 1, 7));
+
+ Mockito.verify(fixture.jobService, Mockito.never())
+ .deleteMappings(Mockito.anyInt(), Mockito.anyList());
+ }
+ }
+
+ @Test
+ public void testDeleteAllowsTerminalTaskAndUsesTransactionalService()
+ throws Exception {
+ Fixture fixture = this.fixture(LoadStatus.SUCCEED);
+
+ fixture.controller.delete("DEFAULT", "hugegraph", 1, 7);
+
+ Mockito.verify(fixture.jobService).deleteMappings(
+ 1, Collections.singletonList(fixture.mapping));
+ Mockito.verify(fixture.mappingService, Mockito.never())
+ .deleteDiskFile(Mockito.any());
+ Mockito.verify(fixture.mappingService, Mockito.never())
+ .remove(Mockito.anyInt());
+ }
+
+ private Fixture fixture(LoadStatus status) throws Exception {
+ Fixture fixture = new Fixture();
+ fixture.controller = new FileMappingController();
+ fixture.mappingService = Mockito.mock(FileMappingService.class);
+ fixture.jobService = Mockito.mock(JobManagerService.class);
+ LoadTaskService taskService = Mockito.mock(LoadTaskService.class);
+ fixture.mapping = new FileMapping();
+ fixture.mapping.setId(7);
+ fixture.mapping.setJobId(1);
+ fixture.mapping.setTotalSize(8L);
+ LoadTask task = LoadTask.builder()
+ .jobId(1)
+ .fileId(7)
+ .status(status)
+ .build();
+ Mockito.when(fixture.mappingService.get("DEFAULT", "hugegraph", 1, 7))
+ .thenReturn(fixture.mapping);
+ Mockito.when(taskService.taskListByJob(1))
+ .thenReturn(Collections.singletonList(task));
+ setField(fixture.controller, "service", fixture.mappingService);
+ setField(fixture.controller, "jobService", fixture.jobService);
+ setField(fixture.controller, "taskService", taskService);
+ return fixture;
+ }
+
+ private static void setField(Object target, String name, Object value)
+ throws Exception {
+ Field field = target.getClass().getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ private static final class Fixture {
+
+ private FileMappingController controller;
+ private FileMappingService mappingService;
+ private JobManagerService jobService;
+ private FileMapping mapping;
+ }
+}
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingSchemaTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingSchemaTest.java
index 0327cae3c..40a922d6b 100644
--- a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingSchemaTest.java
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileMappingSchemaTest.java
@@ -25,7 +25,10 @@ import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Statement;
+import java.util.HashSet;
+import java.util.Set;
import org.junit.Test;
import org.springframework.core.io.FileSystemResource;
@@ -126,6 +129,279 @@ public class FileMappingSchemaTest {
}
}
+ @Test
+ public void testSchemaMigratorAddsMissingLegacyColumns() throws Exception {
+ String url = "jdbc:h2:mem:legacy_schema_columns;DB_CLOSE_DELAY=-1";
+ try (Connection conn = DriverManager.getConnection(url)) {
+ ScriptUtils.executeSqlScript(conn, new FileSystemResource(
+ this.legacySchemaPath()));
+ try (Statement statement = conn.createStatement()) {
+ statement.execute("INSERT INTO `graph_connection` " +
+ "(`name`, `graph`, `host`, `port`, " +
+ "`create_time`) VALUES " +
+ "('legacy', 'legacygraph', 'localhost', " +
+ "8080, CURRENT_TIMESTAMP)");
+ statement.execute("INSERT INTO `execute_history` " +
+ "(`execute_type`, `content`, " +
+ "`execute_status`, `duration`, " +
+ "`create_time`) VALUES " +
+ "(1, 'g.V()', 1, 10, CURRENT_TIMESTAMP)");
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`name`, `content`, `create_time`) " +
+ "VALUES ('saved', 'g.E()', " +
+ "CURRENT_TIMESTAMP)");
+ }
+
+ DatabaseSchemaMigrator migrator = new DatabaseSchemaMigrator();
+ // Running twice must be a no-op the second time
+ migrator.migrate(conn);
+ migrator.migrate(conn);
+
+ for (String column : new String[]{"CONN_ID", "GRAPHSPACE",
+ "GRAPH", "ASYNC_ID", "TEXT",
+ "ASYNC_STATUS",
+ "FAILURE_REASON"}) {
+ this.assertColumnExists(conn, "EXECUTE_HISTORY", column);
+ }
+ for (String column : new String[]{"CONN_ID", "GRAPHSPACE",
+ "GRAPH", "TYPE"}) {
+ this.assertColumnExists(conn, "GREMLIN_COLLECTION", column);
+ }
+
+ // The legacy rows survive and the new columns are backfilled
+ try (Statement statement = conn.createStatement();
+ ResultSet rs = statement.executeQuery(
+ "SELECT `content`, `conn_id`, `graphspace`, `graph`, " +
+ "`async_id`, `text`, `async_status` " +
+ "FROM `execute_history`")) {
+ Assert.assertTrue(rs.next());
+ Assert.assertEquals("g.V()", rs.getString(1));
+ Assert.assertNull(rs.getObject(2));
+ Assert.assertEquals("DEFAULT", rs.getString(3));
+ Assert.assertEquals("legacygraph", rs.getString(4));
+ Assert.assertEquals(0L, rs.getLong(5));
+ Assert.assertEquals("", rs.getString(6));
+ Assert.assertEquals(0, rs.getInt(7));
+ }
+ try (Statement statement = conn.createStatement();
+ ResultSet rs = statement.executeQuery(
+ "SELECT `content`, `conn_id`, `graphspace`, `graph`, " +
+ "`type` " +
+ "FROM `gremlin_collection`")) {
+ Assert.assertTrue(rs.next());
+ Assert.assertEquals("g.E()", rs.getString(1));
+ Assert.assertNull(rs.getObject(2));
+ Assert.assertEquals("DEFAULT", rs.getString(3));
+ Assert.assertEquals("legacygraph", rs.getString(4));
+ Assert.assertEquals("GREMLIN", rs.getString(5));
+ }
+
+ // Names are unique only inside the application-visible scope.
+ try (Statement statement = conn.createStatement()) {
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`conn_id`, `graphspace`, `graph`, `name`, " +
+ "`type`, `content`, `create_time`) VALUES " +
+ "(1, 'DEFAULT', 'othergraph', 'saved', " +
+ "'GREMLIN', 'g.V()', CURRENT_TIMESTAMP)");
+ Assert.assertThrows(SQLException.class, () ->
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`conn_id`, `graphspace`, `graph`, " +
+ "`name`, `type`, `content`, " +
+ "`create_time`) VALUES (1, 'DEFAULT', " +
+ "'legacygraph', 'saved', 'GREMLIN', " +
+ "'g.V()', CURRENT_TIMESTAMP)"));
+ }
+ }
+ }
+
+ @Test
+ public void testSchemaMigratorIsNoopOnFreshSchema() throws Exception {
+ String url = "jdbc:h2:mem:fresh_schema_columns;DB_CLOSE_DELAY=-1";
+ try (Connection conn = DriverManager.getConnection(url)) {
+ ScriptUtils.executeSqlScript(conn, new FileSystemResource(
+ this.mainSchemaPath()));
+
+ DatabaseSchemaMigrator migrator = new DatabaseSchemaMigrator();
+ migrator.migrate(conn);
+ migrator.migrate(conn);
+
+ this.assertColumnExists(conn, "EXECUTE_HISTORY", "GRAPHSPACE");
+ this.assertColumnExists(conn, "GREMLIN_COLLECTION", "TYPE");
+ this.assertIndexExists(conn, "EXECUTE_HISTORY",
+ "EXECUTE_HISTORY_GRAPH_CREATE_TIME");
+ this.assertIndexExists(conn, "FILE_MAPPING",
+ "FILE_MAPPING_JOB_ID");
+ this.assertIndexExists(conn, "LOAD_TASK", "LOAD_TASK_JOB_ID");
+ this.assertIndexExists(conn, "JOB_MANAGER",
+ "JOB_MANAGER_GRAPH_CREATE_TIME");
+ this.assertIndexExists(conn, "ASYNC_TASK", "ASYNC_TASK_GRAPH");
+ // The fresh schema keeps its own column types untouched
+ try (ResultSet columns = conn.getMetaData().getColumns(
+ null, null, "EXECUTE_HISTORY", "GRAPHSPACE")) {
+ Assert.assertTrue(columns.next());
+ Assert.assertEquals(48, columns.getInt("COLUMN_SIZE"));
+ }
+ }
+ }
+
+ @Test
+ public void testSchemaMigratorRecoversScopeFromExistingConnectionId()
+ throws Exception {
+ String url = "jdbc:h2:mem:legacy_schema_conn_id;DB_CLOSE_DELAY=-1";
+ try (Connection conn = DriverManager.getConnection(url);
+ Statement statement = conn.createStatement()) {
+ ScriptUtils.executeSqlScript(conn, new FileSystemResource(
+ this.legacySchemaPath()));
+ statement.execute("INSERT INTO `graph_connection` " +
+ "(`name`, `graph`, `host`, `port`, " +
+ "`create_time`) VALUES " +
+ "('first', 'graph_a', 'host_a', 8080, " +
+ "CURRENT_TIMESTAMP), " +
+ "('second', 'graph_b', 'host_b', 8080, " +
+ "CURRENT_TIMESTAMP)");
+ statement.execute("ALTER TABLE `execute_history` ADD COLUMN " +
+ "`conn_id` INT");
+ statement.execute("ALTER TABLE `execute_history` ADD COLUMN " +
+ "`graphspace` VARCHAR(48)");
+ statement.execute("ALTER TABLE `execute_history` ADD COLUMN " +
+ "`graph` VARCHAR(48)");
+ statement.execute("INSERT INTO `execute_history` " +
+ "(`conn_id`, `graphspace`, `graph`, " +
+ "`execute_type`, `content`, `execute_status`, " +
+ "`duration`, `create_time`) VALUES " +
+ "(2, '', '', 0, 'g.V()', 1, 1, " +
+ "CURRENT_TIMESTAMP)");
+
+ new DatabaseSchemaMigrator().migrate(conn);
+
+ try (ResultSet rows = statement.executeQuery(
+ "SELECT `graphspace`, `graph` FROM `execute_history`")) {
+ Assert.assertTrue(rows.next());
+ Assert.assertEquals("DEFAULT", rows.getString(1));
+ Assert.assertEquals("graph_b", rows.getString(2));
+ }
+ }
+ }
+
+ @Test
+ public void testSchemaMigratorPreservesCollidingLegacyCollections()
+ throws Exception {
+ String url = "jdbc:h2:mem:legacy_collection_collision;DB_CLOSE_DELAY=-1";
+ try (Connection conn = DriverManager.getConnection(url);
+ Statement statement = conn.createStatement()) {
+ statement.execute("CREATE TABLE `graph_connection` (" +
+ "`id` INT NOT NULL AUTO_INCREMENT, " +
+ "`graphspace` VARCHAR(48), " +
+ "`graph` VARCHAR(48), PRIMARY KEY (`id`))");
+ statement.execute("CREATE TABLE `gremlin_collection` (" +
+ "`id` INT NOT NULL AUTO_INCREMENT, " +
+ "`conn_id` INT, `graphspace` VARCHAR(48), " +
+ "`graph` VARCHAR(48), `name` VARCHAR(48), " +
+ "`type` VARCHAR(48), `content` TEXT, " +
+ "PRIMARY KEY (`id`), UNIQUE (`conn_id`, `name`))");
+ statement.execute("INSERT INTO `graph_connection` " +
+ "(`graphspace`, `graph`) VALUES " +
+ "('DEFAULT', 'shared'), ('DEFAULT', 'shared')");
+ statement.execute("INSERT INTO `gremlin_collection` " +
+ "(`conn_id`, `name`, `content`) VALUES " +
+ "(NULL, 'saved', 'first'), " +
+ "(NULL, 'saved', 'second'), " +
+ "(NULL, 'saved_2', 'reserved')");
+
+ DatabaseSchemaMigrator migrator = new DatabaseSchemaMigrator();
+ migrator.migrate(conn);
+ migrator.migrate(conn);
+
+ Set<String> names = new HashSet<>();
+ Set<String> contents = new HashSet<>();
+ try (ResultSet rows = statement.executeQuery(
+ "SELECT `conn_id`, `name`, `content` " +
+ "FROM `gremlin_collection`")) {
+ while (rows.next()) {
+ Assert.assertNull(rows.getObject(1));
+ names.add(rows.getString(2));
+ contents.add(rows.getString(3));
+ }
+ }
+ Assert.assertEquals(3, names.size());
+ Assert.assertTrue(names.contains("saved"));
+ Assert.assertTrue(names.contains("saved_2"));
+ Assert.assertTrue(names.stream().anyMatch(
+ name -> name.startsWith("saved_2_") &&
+ !name.equals("saved_2")));
+ Assert.assertEquals(3, contents.size());
+ Assert.assertTrue(contents.contains("first"));
+ Assert.assertTrue(contents.contains("second"));
+ Assert.assertTrue(contents.contains("reserved"));
+ }
+ }
+
+ @Test
+ public void testSchemaMigratorAddsMissingIndexesIdempotently()
+ throws Exception {
+ String url = "jdbc:h2:mem:legacy_schema_indexes;DB_CLOSE_DELAY=-1";
+ try (Connection conn = DriverManager.getConnection(url);
+ Statement statement = conn.createStatement()) {
+ statement.execute("CREATE TABLE `execute_history` (" +
+ "`id` INT, `graphspace` VARCHAR(48), " +
+ "`graph` VARCHAR(48), `create_time` DATETIME)");
+ statement.execute("CREATE TABLE `file_mapping` (" +
+ "`id` INT, `job_id` INT)");
+ statement.execute("CREATE TABLE `load_task` (" +
+ "`id` INT, `job_id` INT, `options` TEXT)");
+ statement.execute("CREATE TABLE `job_manager` (" +
+ "`id` INT, `graphspace` VARCHAR(48), " +
+ "`graph` VARCHAR(48), `create_time` DATETIME)");
+ statement.execute("CREATE TABLE `async_task` (" +
+ "`id` INT, `graphspace` VARCHAR(48), " +
+ "`graph` VARCHAR(48))");
+
+ DatabaseSchemaMigrator migrator = new DatabaseSchemaMigrator();
+ migrator.migrate(conn);
+ migrator.migrate(conn);
+
+ this.assertIndexExists(conn, "EXECUTE_HISTORY",
+ "EXECUTE_HISTORY_GRAPH_CREATE_TIME");
+ this.assertIndexExists(conn, "FILE_MAPPING",
+ "FILE_MAPPING_JOB_ID");
+ this.assertIndexExists(conn, "LOAD_TASK", "LOAD_TASK_JOB_ID");
+ this.assertIndexExists(conn, "JOB_MANAGER",
+ "JOB_MANAGER_GRAPH_CREATE_TIME");
+ this.assertIndexExists(conn, "ASYNC_TASK", "ASYNC_TASK_GRAPH");
+ }
+ }
+
+ private void assertColumnExists(Connection conn, String table,
+ String column) throws Exception {
+ try (ResultSet columns = conn.getMetaData().getColumns(
+ null, null, table, column)) {
+ Assert.assertTrue(table + "." + column + " should exist",
+ columns.next());
+ }
+ }
+
+ private void assertIndexExists(Connection conn, String table,
+ String index) throws Exception {
+ try (ResultSet indexes = conn.getMetaData().getIndexInfo(
+ conn.getCatalog(), null, table, false, false)) {
+ while (indexes.next()) {
+ if (index.equalsIgnoreCase(indexes.getString("INDEX_NAME"))) {
+ return;
+ }
+ }
+ }
+ Assert.fail(table + "." + index + " should exist");
+ }
+
+ private Path legacySchemaPath() {
+ Path modulePath = Paths.get("src/test/resources/database/schema.sql");
+ if (Files.exists(modulePath)) {
+ return modulePath;
+ }
+ return Paths.get("hugegraph-hubble/hubble-be/src/test/resources/" +
+ "database/schema.sql");
+ }
+
private void insertDeepPath(Connection conn, String deepPath)
throws Exception {
try (PreparedStatement insert = conn.prepareStatement(
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileUploadControllerTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileUploadControllerTest.java
index d6b251789..269d71d07 100644
--- a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileUploadControllerTest.java
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/FileUploadControllerTest.java
@@ -240,6 +240,55 @@ public class FileUploadControllerTest {
}
}
+ @Test
+ public void testRestoreMovedUploadRestoresFileAndRemovesDirectory()
+ throws Exception {
+ Path uploadRoot = Files.createTempDirectory("hubble-upload-restore");
+ try {
+ Path movedDir = Files.createDirectories(
+ uploadRoot.resolve("file-mapping-2"));
+ Path movedFile = Files.write(movedDir.resolve("data.csv"),
+ "id,name".getBytes());
+ Path originalFile = uploadRoot.resolve("data.csv");
+ FileMappingService service = this.fileMappingService(uploadRoot);
+
+ service.restoreMovedUpload(movedFile.toString(),
+ originalFile.toString());
+
+ Assert.assertTrue(Files.exists(originalFile));
+ Assert.assertFalse(Files.exists(movedFile));
+ Assert.assertFalse(Files.exists(movedDir));
+ } finally {
+ FileUtils.deleteDirectory(uploadRoot.toFile());
+ }
+ }
+
+ @Test
+ public void testRestoreMovedUploadRejectsSymlinkedDestinationParent()
+ throws Exception {
+ Path uploadRoot = Files.createTempDirectory("hubble-upload-root");
+ Path outsideRoot = Files.createTempDirectory("hubble-upload-outside");
+ try {
+ Path movedDir = Files.createDirectories(
+ uploadRoot.resolve("file-mapping-2"));
+ Path movedFile = Files.write(movedDir.resolve("data.csv"),
+ "id,name".getBytes());
+ Path linkedParent = uploadRoot.resolve("linked");
+ Files.createSymbolicLink(linkedParent, outsideRoot);
+ Path originalFile = linkedParent.resolve("data.csv");
+ FileMappingService service = this.fileMappingService(uploadRoot);
+
+ Assert.assertThrows(InternalException.class, () ->
+ service.restoreMovedUpload(movedFile.toString(),
+ originalFile.toString()));
+ Assert.assertTrue(Files.exists(movedFile));
+ Assert.assertFalse(Files.exists(outsideRoot.resolve("data.csv")));
+ } finally {
+ FileUtils.deleteDirectory(uploadRoot.toFile());
+ FileUtils.deleteDirectory(outsideRoot.toFile());
+ }
+ }
+
@Test
public void testTryMergePartFilesReplacesStaleAllFile() throws Exception {
Path parent = Files.createTempDirectory("hubble-merge");
@@ -262,6 +311,78 @@ public class FileUploadControllerTest {
}
}
+ @Test
+ public void testFinalizeUploadRestoresMovedFileAfterDatabaseFailure()
+ throws Exception {
+ FileUploadController controller = this.controller("csv");
+ FileMappingService service = Mockito.mock(FileMappingService.class);
+ JobManagerService jobService = Mockito.mock(JobManagerService.class);
+ FileMapping mapping = this.completedMapping("upload/data.csv");
+ Mockito.when(service.moveToNextLevelDir(mapping))
+ .thenReturn("upload/file-mapping-2/data.csv");
+ Mockito.doThrow(new InternalException("database failure"))
+ .when(jobService).completeUpload(mapping);
+ this.setField(controller, "service", service);
+ this.setField(controller, "jobService", jobService);
+
+ Assert.assertThrows(InternalException.class,
+ () -> this.finalizeUpload(controller, mapping));
+
+ Mockito.verify(service).restoreMovedUpload(
+ "upload/file-mapping-2/data.csv", "upload/data.csv");
+ Mockito.verify(service, Mockito.never()).update(mapping);
+ Assert.assertEquals("upload/data.csv", mapping.getPath());
+ Assert.assertEquals(FileMappingStatus.UPLOADING,
+ mapping.getFileStatus());
+ }
+
+ @Test
+ public void testFinalizeUploadRetainsCleanupPathWhenRestoreFails()
+ throws Exception {
+ FileUploadController controller = this.controller("csv");
+ FileMappingService service = Mockito.mock(FileMappingService.class);
+ JobManagerService jobService = Mockito.mock(JobManagerService.class);
+ FileMapping mapping = this.completedMapping("upload/data.csv");
+ String movedPath = "upload/file-mapping-2/data.csv";
+ Mockito.when(service.moveToNextLevelDir(mapping)).thenReturn(movedPath);
+ Mockito.doThrow(new InternalException("database failure"))
+ .when(jobService).completeUpload(mapping);
+ Mockito.doThrow(new InternalException("restore failure"))
+ .when(service).restoreMovedUpload(movedPath,
+ "upload/data.csv");
+ this.setField(controller, "service", service);
+ this.setField(controller, "jobService", jobService);
+
+ Assert.assertThrows(InternalException.class,
+ () -> this.finalizeUpload(controller, mapping));
+
+ Mockito.verify(service).update(mapping);
+ Assert.assertEquals(movedPath, mapping.getPath());
+ Assert.assertEquals(FileMappingStatus.FAILURE,
+ mapping.getFileStatus());
+
+ JobManager job = JobManager.builder().id(1).jobSize(0L).build();
+ Mockito.when(jobService.get("DEFAULT", "hugegraph", 1))
+ .thenReturn(job);
+ Mockito.when(service.get("DEFAULT", "hugegraph", 1, "data.csv"))
+ .thenReturn(mapping);
+ Assert.assertThrows(ExternalException.class, () ->
+ this.reserveUploadQuota(controller, "DEFAULT", "hugegraph",
+ 1, "data.csv", "upload/data.csv",
+ 10L));
+ Mockito.verify(service).update(mapping);
+ }
+
+ private FileMapping completedMapping(String path) {
+ FileMapping mapping = new FileMapping("DEFAULT", "hugegraph",
+ "data.csv", path);
+ mapping.setId(2);
+ mapping.setJobId(1);
+ mapping.setFileStatus(FileMappingStatus.COMPLETED);
+ mapping.setTotalSize(10L);
+ return mapping;
+ }
+
private FileUploadController controller(String... formats) throws Exception {
FileUploadController controller = new FileUploadController();
HugeConfig config = Mockito.mock(HugeConfig.class);
@@ -275,6 +396,16 @@ public class FileUploadControllerTest {
return controller;
}
+ private FileMappingService fileMappingService(Path uploadRoot)
+ throws Exception {
+ FileMappingService service = new FileMappingService();
+ HugeConfig config = Mockito.mock(HugeConfig.class);
+ Mockito.when(config.get(HubbleOptions.UPLOAD_FILE_LOCATION))
+ .thenReturn(uploadRoot.toString());
+ this.setField(service, "config", config);
+ return service;
+ }
+
private void checkFileValid(FileUploadController controller, JobManager job,
MockMultipartFile file, String fileName)
throws Exception {
@@ -321,6 +452,22 @@ public class FileUploadControllerTest {
}
}
+ private void finalizeUpload(FileUploadController controller,
+ FileMapping mapping) throws Exception {
+ Method method = FileUploadController.class.getDeclaredMethod(
+ "finalizeUpload", FileMapping.class);
+ method.setAccessible(true);
+ try {
+ method.invoke(controller, mapping);
+ } catch (InvocationTargetException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ throw e;
+ }
+ }
+
private void checkTotalAndIndexValid(FileUploadController controller,
int total, int index)
throws Exception {
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/GremlinUtilTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/GremlinUtilTest.java
new file mode 100644
index 000000000..d4cdfc60c
--- /dev/null
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/GremlinUtilTest.java
@@ -0,0 +1,190 @@
+/*
+ *
+ * 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 org.apache.hugegraph.unit;
+
+import org.junit.Test;
+
+import org.apache.hugegraph.testutil.Assert;
+import org.apache.hugegraph.util.GremlinUtil;
+
+public class GremlinUtilTest {
+
+ private static final int LIMIT = 250;
+
+ @Test
+ public void testSuffixAppend() {
+ // Matched suffixes get '.limit(N)' appended
+ Assert.assertEquals("g.V().limit(250)",
+ GremlinUtil.optimizeLimit("g.V()", LIMIT));
+ Assert.assertEquals("g.V().out().limit(250)",
+ GremlinUtil.optimizeLimit("g.V().out()", LIMIT));
+ Assert.assertEquals("g.E().limit(250)",
+ GremlinUtil.optimizeLimit("g.E()", LIMIT));
+ Assert.assertEquals("g.V().hasLabel('person').limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().hasLabel('person')", LIMIT));
+ // Unmatched suffixes are left untouched
+ Assert.assertEquals("g.V().count()",
+ GremlinUtil.optimizeLimit("g.V().count()", LIMIT));
+ // Multi-line: each line is optimized independently
+ Assert.assertEquals("g.V().limit(250)\ng.V().count()",
+ GremlinUtil.optimizeLimit(
+ "g.V()\ng.V().count()", LIMIT));
+ }
+
+ @Test
+ public void testMultilineChainReceivesOnlyOneTailLimit() {
+ Assert.assertEquals("g.V()\n .hasLabel(\"person\")\n " +
+ ".out().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V()\n .hasLabel(\"person\")\n .out()",
+ LIMIT));
+ Assert.assertEquals("g.V() // continue\n .out().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V() // continue\n .out()", LIMIT));
+ Assert.assertEquals("g.V()\n/* keep traversing */\n" +
+ ".out().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V()\n/* keep traversing */\n.out()",
+ LIMIT));
+ }
+
+ @Test
+ public void testApostropheInCommentDoesNotDisableLimit() {
+ // An apostrophe inside a comment must not be read as a string
+ // delimiter, otherwise the rest of the script is treated as one
+ // unterminated string and never receives its limit.
+ String optimized = GremlinUtil.optimizeLimit(
+ "// don't limit this\ng.V()", 250);
+ Assert.assertTrue(optimized, optimized.endsWith(".limit(250)"));
+ Assert.assertTrue(optimized, optimized.startsWith("// don't limit"));
+ }
+
+ @Test
+ public void testQuoteInCommentDoesNotLeakIntoNextStatement() {
+ String optimized = GremlinUtil.optimizeLimit(
+ "// it's here\ng.V();g.E()", 10);
+ Assert.assertEquals("// it's here\ng.V().limit(10);g.E().limit(10)",
+ optimized);
+ }
+
+ @Test
+ public void testIgnoredCommentLine() {
+ Assert.assertEquals("// g.V()\ng.E().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "// g.V()\ng.E()", LIMIT));
+ }
+
+ @Test
+ public void testTrailingLineCommentKeepsLimitBeforeComment() {
+ Assert.assertEquals("g.V().limit(250) // trailing",
+ GremlinUtil.optimizeLimit(
+ "g.V() // trailing", LIMIT));
+ }
+
+ @Test
+ public void testBlockCommentDoesNotAffectStatementParsing() {
+ Assert.assertEquals("g.V().limit(250) /* '; \\\" */;" +
+ "g.E().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V() /* '; \\\" */;g.E()", LIMIT));
+ Assert.assertEquals("g.V().limit(250) /* line one\n" +
+ "line two */\ng.E().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V() /* line one\nline two */\ng.E()", LIMIT));
+ }
+
+ @Test
+ public void testPlainMultiStatement() {
+ Assert.assertEquals("g.V().limit(250);g.E().limit(250)",
+ GremlinUtil.optimizeLimit("g.V();g.E()", LIMIT));
+ Assert.assertEquals("g.V().count();g.E().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().count();g.E()", LIMIT));
+ // Trailing separator kept, no limit appended to the empty tail
+ Assert.assertEquals("g.V().limit(250);",
+ GremlinUtil.optimizeLimit("g.V();", LIMIT));
+ }
+
+ @Test
+ public void testSemicolonInsideSingleQuotedString() {
+ // The ';' inside the quoted string is not a statement separator
+ Assert.assertEquals("g.V().hasLabel('a;b').limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().hasLabel('a;b')", LIMIT));
+ Assert.assertEquals("g.V().hasLabel('a;b').limit(250);" +
+ "g.E().limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().hasLabel('a;b');g.E()", LIMIT));
+ }
+
+ @Test
+ public void testSemicolonInsideDoubleQuotedString() {
+ Assert.assertEquals("g.V().hasLabel(\"a;b\").limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().hasLabel(\"a;b\")", LIMIT));
+ }
+
+ @Test
+ public void testEscapedQuoteInsideString() {
+ // The escaped quote does not close the string, so the ';' and the
+ // fake suffix inside it are ignored
+ Assert.assertEquals("g.V().hasLabel('it\\'s;x').limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().hasLabel('it\\'s;x')", LIMIT));
+ Assert.assertEquals("g.V().hasLabel(\"say \\\";\\\" ok\").limit(250)",
+ GremlinUtil.optimizeLimit(
+ "g.V().hasLabel(\"say \\\";\\\" ok\")", LIMIT));
+ }
+
+ @Test
+ public void testNewlineInsideStringNotSeparator() {
+ // A '\n' inside quotes must not be treated as a line separator
+ String gremlin = "g.V().hasLabel('a\nb')";
+ Assert.assertEquals("g.V().hasLabel('a\nb').limit(250)",
+ GremlinUtil.optimizeLimit(gremlin, LIMIT));
+ }
+
+ @Test
+ public void testUnclosedQuoteNotOptimized() {
+ // Never append '.limit(N)' inside an unterminated string literal
+ String gremlin = "g.V().hasLabel('unclosed";
+ Assert.assertEquals(gremlin,
+ GremlinUtil.optimizeLimit(gremlin, LIMIT));
+ }
+
+ @Test
+ public void testUnsupportedGroovyLiteralsAreNotOptimized() {
+ String slashy = "def x = /g.V()\\n/\ng.E()";
+ Assert.assertEquals(slashy,
+ GremlinUtil.optimizeLimit(slashy, LIMIT));
+
+ String dollarSlashy = "def x = $/g.V()/$\ng.E()";
+ Assert.assertEquals(dollarSlashy,
+ GremlinUtil.optimizeLimit(dollarSlashy, LIMIT));
+
+ String tripleQuoted = "def x = '''g.V()\ng.E()'''\ng.V()";
+ Assert.assertEquals(tripleQuoted,
+ GremlinUtil.optimizeLimit(tripleQuoted, LIMIT));
+
+ String division = "def half = 4 / 2\ng.V()";
+ Assert.assertEquals(division,
+ GremlinUtil.optimizeLimit(division, LIMIT));
+ }
+}
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/LoaderScopeControllerTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/LoaderScopeControllerTest.java
index 3a918e49d..e8f6f038f 100644
--- a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/LoaderScopeControllerTest.java
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/LoaderScopeControllerTest.java
@@ -78,15 +78,24 @@ public class LoaderScopeControllerTest {
@Test
public void testFileClearUsesNestedScope() {
FileMappingService service = Mockito.mock(FileMappingService.class);
+ JobManagerService jobService = Mockito.mock(JobManagerService.class);
+ LoadTaskService taskService = Mockito.mock(LoadTaskService.class);
FileMapping mapping = FileMapping.builder().id(11).build();
Mockito.when(service.listByJob("space-a", "graph-a", 7))
.thenReturn(Collections.singletonList(mapping));
+ Mockito.when(taskService.taskListByJob(7))
+ .thenReturn(Collections.emptyList());
FileMappingController controller = new FileMappingController();
ReflectionTestUtils.setField(controller, "service", service);
+ ReflectionTestUtils.setField(controller, "jobService", jobService);
+ ReflectionTestUtils.setField(controller, "taskService", taskService);
controller.clear("space-a", "graph-a", 7);
- Mockito.verify(service).remove(11);
+ Mockito.verify(jobService).deleteMappings(
+ 7, Collections.singletonList(mapping));
+ Mockito.verify(service, Mockito.never()).deleteDiskFile(mapping);
+ Mockito.verify(service, Mockito.never()).remove(11);
Mockito.verify(service, Mockito.never()).listAll();
}
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/PriorityFixTest.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/PriorityFixTest.java
index 5581c410a..b30bfd4ee 100644
--- a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/PriorityFixTest.java
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/PriorityFixTest.java
@@ -73,6 +73,16 @@ public class PriorityFixTest {
PageUtil.checkPositivePage(1, -1);
}
+ @Test(expected = IllegalArgumentException.class)
+ public void testBoundedIntRejectsUnsupportedNegativePageSize() {
+ PageUtil.boundedSize(-2);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testBoundedLongRejectsUnsupportedNegativePageSize() {
+ PageUtil.boundedSize(-2L);
+ }
+
@Test(expected = ExternalException.class)
public void testInterceptorRejectsOversizedPage() throws Exception {
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/api/list");
@@ -83,6 +93,16 @@ public class PriorityFixTest {
new Object());
}
+ @Test(expected = ExternalException.class)
+ public void testInterceptorRejectsUnsupportedNegativePageSize() throws Exception {
+ MockHttpServletRequest request = new MockHttpServletRequest("GET", "/api/list");
+ request.setParameter("page_size", "-2");
+
+ new CustomInterceptor().preHandle(request,
+ new MockHttpServletResponse(),
+ new Object());
+ }
+
@Test
public void testInterceptorKeepsLegacyAllPageSentinel() throws Exception {
MockHttpServletRequest request = new MockHttpServletRequest("GET", "/api/list");
diff --git a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/UnitTestSuite.java
index 6d8f611db..a6b487b09 100644
--- a/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/UnitTestSuite.java
+++ b/hugegraph-hubble/hubble-be/src/test/java/org/apache/hugegraph/unit/UnitTestSuite.java
@@ -60,6 +60,7 @@ import org.junit.runners.Suite;
GraphSpaceAuthOwnershipTest.class,
GraphSpaceServiceTest.class,
GraphsControllerCanonicalTest.class,
+ GremlinUtilTest.class,
GremlinHistoryFailureTest.class,
HubbleOptionsTest.class,
IngestControllerTest.class,
diff --git a/hugegraph-hubble/hubble-dist/assembly/static/bin/start-hubble.sh b/hugegraph-hubble/hubble-dist/assembly/static/bin/start-hubble.sh
index 6c71becb7..6845fb213 100644
--- a/hugegraph-hubble/hubble-dist/assembly/static/bin/start-hubble.sh
+++ b/hugegraph-hubble/hubble-dist/assembly/static/bin/start-hubble.sh
@@ -28,6 +28,15 @@ LIB_PATH=${HOME_PATH}/lib
LOG_PATH=${HOME_PATH}/logs
PID_FILE=${BIN_PATH}/pid
+process_start_time() {
+ local process_pid=$1
+ if [[ -r /proc/${process_pid}/stat ]]; then
+ awk '{print $22}' "/proc/${process_pid}/stat"
+ else
+ LC_ALL=C ps -o lstart= -p "${process_pid}" 2>/dev/null
+ fi
+}
+
. "${BIN_PATH}"/common_functions
java_env_check
@@ -74,17 +83,36 @@ while [[ $# -gt 0 ]]; do
shift
done
+MAIN_CLASS="org.apache.hugegraph.HugeGraphHubble"
+
if [[ -f ${PID_FILE} ]] ; then
- PID=$(cat "${PID_FILE}")
- if kill -0 "${PID}" > /dev/null 2>&1; then
- echo "HugeGraphHubble is running as process ${PID}, please stop it first!"
- exit 1
+ read -r PID PID_START < "${PID_FILE}"
+ if [[ ! ${PID} =~ ^[0-9]+$ ]]; then
+ echo "Invalid HugeGraphHubble PID file, removing it"
+ rm "${PID_FILE}"
+ elif kill -0 "${PID}" > /dev/null 2>&1; then
+ CURRENT_START=$(process_start_time "${PID}") || CURRENT_START=""
+ if [[ -z ${PID_START} ]]; then
+ PROCESS_ARGS=$(ps -p "${PID}" -o args= 2>/dev/null || true)
+ if [[ ${PROCESS_ARGS} == *"${MAIN_CLASS}"* &&
+ ${PROCESS_ARGS} == *"-Dhubble.home.path=${HOME_PATH}"* ]]; then
+ echo "HugeGraphHubble is running as process ${PID}, please stop it first!"
+ exit 1
+ fi
+ echo "Stale HugeGraphHubble PID file, removing it"
+ rm "${PID_FILE}"
+ elif [[ ${PID_START} != "${CURRENT_START}" ]]; then
+ echo "Stale HugeGraphHubble PID file, removing it"
+ rm "${PID_FILE}"
+ else
+ echo "HugeGraphHubble is running as process ${PID}, please stop it first!"
+ exit 1
+ fi
else
rm "${PID_FILE}"
fi
fi
-MAIN_CLASS="org.apache.hugegraph.HugeGraphHubble"
ARGS=${CONF_PATH}/hugegraph-hubble.properties
LOG=${LOG_PATH}/hugegraph-hubble.log
@@ -99,7 +127,13 @@ else
fi
PID=$!
-echo ${PID} > "${PID_FILE}"
+PID_START=$(process_start_time "${PID}") || PID_START=""
+if ! kill -0 "${PID}" > /dev/null 2>&1; then
+ wait "${PID}" || true
+ cat "${LOG}" || true
+ exit 1
+fi
+echo "${PID} ${PID_START}" > "${PID_FILE}"
# wait hubble start
TIMEOUT_S=30
diff --git a/hugegraph-hubble/hubble-dist/assembly/static/bin/stop-hubble.sh b/hugegraph-hubble/hubble-dist/assembly/static/bin/stop-hubble.sh
index 03f1015af..8db837359 100644
--- a/hugegraph-hubble/hubble-dist/assembly/static/bin/stop-hubble.sh
+++ b/hugegraph-hubble/hubble-dist/assembly/static/bin/stop-hubble.sh
@@ -19,13 +19,77 @@ HOME_PATH=$(dirname "$0")
HOME_PATH=$(cd "${HOME_PATH}"/.. && pwd)
BIN_PATH=${HOME_PATH}/bin
PID_FILE=${BIN_PATH}/pid
+# Seconds to wait for a graceful shutdown before resorting to SIGKILL
+STOP_TIMEOUT=${STOP_TIMEOUT:-30}
+
+if [[ ! ${STOP_TIMEOUT} =~ ^[0-9]+$ ]]; then
+ echo "STOP_TIMEOUT must be a non-negative integer" >&2
+ exit 1
+fi
+
+process_start_time() {
+ local process_pid=$1
+ if [[ -r /proc/${process_pid}/stat ]]; then
+ awk '{print $22}' "/proc/${process_pid}/stat"
+ else
+ LC_ALL=C ps -o lstart= -p "${process_pid}" 2>/dev/null
+ fi
+}
if [[ -f ${PID_FILE} ]]; then
- pid=$(cat "${PID_FILE}")
- if kill -0 "${pid}" > /dev/null 2>&1; then
- kill -9 "${pid}"
+ read -r pid expected_start < "${PID_FILE}"
+ if [[ ! ${pid} =~ ^[0-9]+$ ]]; then
+ echo "Invalid HugeGraphHubble PID file; refusing to signal a process" >&2
+ exit 1
+ fi
+
+ same_process() {
+ if ! kill -0 "${pid}" > /dev/null 2>&1; then
+ return 1
+ fi
+ if [[ -n ${expected_start} ]]; then
+ [[ $(process_start_time "${pid}") == "${expected_start}" ]]
+ return
+ fi
+ # Backward compatibility for PID files written by older launchers.
+ process_args=$(ps -p "${pid}" -o args= 2>/dev/null) || return 1
+ [[ ${process_args} == *"org.apache.hugegraph.HugeGraphHubble"* &&
+ ${process_args} == *"-Dhubble.home.path=${HOME_PATH}"* ]]
+ }
+
+ if same_process; then
+ # SIGTERM first so the Spring shutdown hooks run: they pause the
+ # running load tasks and close the H2 database cleanly. Only escalate
+ # to SIGKILL if the process is still alive after STOP_TIMEOUT.
+ kill "${pid}" > /dev/null 2>&1
+ waited=0
+ while [ "${waited}" -lt "${STOP_TIMEOUT}" ]; do
+ if ! same_process; then
+ break
+ fi
+ sleep 1
+ waited=$((waited + 1))
+ done
+ if same_process; then
+ echo "HugeGraphHubble did not exit within ${STOP_TIMEOUT}s, killing it"
+ kill -9 "${pid}" > /dev/null 2>&1
+ fi
+ for _ in 1 2 3 4 5; do
+ if ! same_process; then
+ break
+ fi
+ sleep 1
+ done
+ if same_process; then
+ echo "HugeGraphHubble process ${pid} is still alive; retaining PID file" >&2
+ exit 1
+ fi
echo "stopped HugeGraphHubble"
else
+ if kill -0 "${pid}" > /dev/null 2>&1; then
+ echo "PID ${pid} belongs to a different process; retaining PID file" >&2
+ exit 1
+ fi
echo "process ${pid} not exist"
fi
rm "${PID_FILE}"
diff --git a/hugegraph-hubble/hubble-fe/src/api/request.js b/hugegraph-hubble/hubble-fe/src/api/request.js
index 8ce3355fc..84a534b6a 100644
--- a/hugegraph-hubble/hubble-fe/src/api/request.js
+++ b/hugegraph-hubble/hubble-fe/src/api/request.js
@@ -88,9 +88,26 @@ const notifyForbidden = config => {
}
};
+// Axios buildURL un-escapes [ and ] in query strings, but Tomcat rejects raw
+// brackets in the request target with a plain 400, so encode every character.
+const serializeParams = params => {
+ const parts = [];
+ Object.entries(params || {}).forEach(([key, value]) => {
+ if (value === undefined || value === null) {
+ return;
+ }
+ const values = Array.isArray(value) ? value : [value];
+ values.forEach(v => {
+ parts.push(encodeURIComponent(key) + '=' + encodeURIComponent(v));
+ });
+ });
+ return parts.join('&');
+};
+
const instance = axios.create({
baseURL: '/api/v1.3',
withCredentials: true,
+ paramsSerializer: serializeParams,
// Backend times out after 30s; keep this slightly higher to receive its error body.
timeout: 31000,
transformResponse: [parseResponse],
diff --git a/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/analysis.json b/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/analysis.json
index cea1beb88..03c799cd3 100644
--- a/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/analysis.json
+++ b/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/analysis.json
@@ -80,6 +80,7 @@
"mode_2d": "2D Mode",
"mode_3d": "3D Mode",
"import_success": "Imported successfully",
+ "import_failed": "Import failed",
"export_json": "Export JSON",
"export_png": "Export image",
"export_json_title": "Export JSON",
diff --git a/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/pages.json b/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/pages.json
index d00fd29ed..6f55f7414 100644
--- a/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/pages.json
+++ b/hugegraph-hubble/hubble-fe/src/i18n/resources/en-US/modules/pages.json
@@ -576,6 +576,7 @@
"sync_type": "Sync Type",
"status": "Status",
"create_time": "Created At",
+ "creator": "Creator",
"operation": "Actions"
},
"statistic": {
@@ -988,6 +989,7 @@
"delete_confirm": "Confirm deleting this edge type?",
"delete_warning": "This action cannot be undone. Please proceed with caution.",
"delete_task_note": "Deleting metadata may take a while. Check Async Tasks for progress.",
+ "parent_hint": "Parent: {{name}}",
"col": {
"name": "Edge Type Name",
"type": "Edge Type",
diff --git a/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/analysis.json b/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/analysis.json
index f2117218a..13ddc2b95 100644
--- a/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/analysis.json
+++ b/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/analysis.json
@@ -80,6 +80,7 @@
"mode_2d": "2D 模式",
"mode_3d": "3D 模式",
"import_success": "导入成功",
+ "import_failed": "导入失败",
"export_json": "导出 JSON",
"export_png": "导出图片",
"export_json_title": "导出 JSON",
diff --git a/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/pages.json b/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/pages.json
index 2737959ce..a80f7f0d7 100644
--- a/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/pages.json
+++ b/hugegraph-hubble/hubble-fe/src/i18n/resources/zh-CN/modules/pages.json
@@ -576,6 +576,7 @@
"sync_type": "同步类型",
"status": "状态",
"create_time": "创建时间",
+ "creator": "创建人",
"operation": "操作"
},
"statistic": {
@@ -988,6 +989,7 @@
"delete_confirm": "确认删除此边类型？",
"delete_warning": "删除后无法恢复，请谨慎操作",
"delete_task_note": "删除元数据耗时较久，详情可在异步任务中查看",
+ "parent_hint": "父边：{{name}}",
"col": {
"name": "边类型名称",
"type": "边类型",
diff --git a/hugegraph-hubble/hubble-fe/src/modules/algorithm/Home/index.js b/hugegraph-hubble/hubble-fe/src/modules/algorithm/Home/index.js
index 88cd565aa..c9fae69db 100644
--- a/hugegraph-hubble/hubble-fe/src/modules/algorithm/Home/index.js
+++ b/hugegraph-hubble/hubble-fe/src/modules/algorithm/Home/index.js
@@ -134,6 +134,13 @@ const AlgorithmHome = () => {
const getFavoriteQueriesList = useCallback(
async () => {
+ if (!graphSpace || !graph) {
+ favoriteQueriesRequest.current = null;
+ setFavoriteQueriesLoading(false);
+ setFavoriteQueriesError(false);
+ setFavoriteQueriesData({});
+ return;
+ }
const request = Symbol('algorithm-favorites');
favoriteQueriesRequest.current = request;
const params = {
@@ -178,6 +185,13 @@ const AlgorithmHome = () => {
const getExecutionLogsList = useCallback(
async () => {
+ if (!graphSpace || !graph) {
+ executionLogsRequest.current = null;
+ setExecutionLogsLoading(false);
+ setExecutionLogsError(false);
+ setExecutionLogsData({});
+ return;
+ }
const request = Symbol('algorithm-logs');
executionLogsRequest.current = request;
const params = {'page_size': pageSize, 'page_no': pageExecute, 'type': TYPE.ALGORITHM};
diff --git a/hugegraph-hubble/hubble-fe/src/modules/analysis/Home/index.js b/hugegraph-hubble/hubble-fe/src/modules/analysis/Home/index.js
index 326f68008..dec24f9ba 100644
--- a/hugegraph-hubble/hubble-fe/src/modules/analysis/Home/index.js
+++ b/hugegraph-hubble/hubble-fe/src/modules/analysis/Home/index.js
@@ -116,7 +116,11 @@ const AnalysisHome = () => {
const getExecutionLogsList = useCallback(
async () => {
- if (analysisMode === TEXT2GQL) {
+ if (analysisMode === TEXT2GQL || !graphSpace || !graph) {
+ executionLogsRequest.current = null;
+ setExecutionLogsLoading(false);
+ setExecutionLogsError(false);
+ setExecutionLogsData({});
return;
}
const request = Symbol('execution-logs');
@@ -154,7 +158,11 @@ const AnalysisHome = () => {
const getFavoriteQueriesList = useCallback(
async () => {
- if (analysisMode === TEXT2GQL) {
+ if (analysisMode === TEXT2GQL || !graphSpace || !graph) {
+ favoriteQueriesRequest.current = null;
+ setFavoriteQueriesLoading(false);
+ setFavoriteQueriesError(false);
+ setFavoriteQueriesData({});
return;
}
const request = Symbol('favorite-queries');
diff --git a/hugegraph-hubble/hubble-fe/src/modules/asyncTasks/Detail/index.js b/hugegraph-hubble/hubble-fe/src/modules/asyncTasks/Detail/index.js
index e984a7b37..dbe631a5b 100644
--- a/hugegraph-hubble/hubble-fe/src/modules/asyncTasks/Detail/index.js
+++ b/hugegraph-hubble/hubble-fe/src/modules/asyncTasks/Detail/index.js
@@ -215,7 +215,7 @@ const AsyncTaskDetail = props => {
title: t('analysis.async_task.column.create_time'),
dataIndex: 'task_create',
render: (task_create, rowData, index) => {
- const convertedDate = format(new Date(task_create), 'yyyy-MM-dd H:m:ss');
+ const convertedDate = task_create ? format(new Date(task_create), 'yyyy-MM-dd HH:mm:ss') : '--';
return (<>{convertedDate}</>);
},
},
diff --git a/hugegraph-hubble/hubble-fe/src/modules/component/DynamicAddEdge/index.js b/hugegraph-hubble/hubble-fe/src/modules/component/DynamicAddEdge/index.js
index c88205cf1..c4b7a8d40 100644
--- a/hugegraph-hubble/hubble-fe/src/modules/component/DynamicAddEdge/index.js
+++ b/hugegraph-hubble/hubble-fe/src/modules/component/DynamicAddEdge/index.js
@@ -97,8 +97,8 @@ const AddEdgeDrawer = props => {
api.analysis.fetchEdgeLabels(currentGraphSpace, currentGraph, label)
.then(res => {
const {data, status, message: errMsg} = res;
- const {properties} = data;
if (status === 200) {
+ const {properties} = data;
setEdgeProperties(properties);
}
else {
diff --git a/hugegraph-hubble/hubble-fe/src/modules/component/ImportData/index.js b/hugegraph-hubble/hubble-fe/src/modules/component/ImportData/index.js
index 6e6dbab22..a09f7b6ee 100644
--- a/hugegraph-hubble/hubble-fe/src/modules/component/ImportData/index.js
+++ b/hugegraph-hubble/hubble-fe/src/modules/component/ImportData/index.js
@@ -46,17 +46,32 @@ const ImportData = props => {
const handleUploadChange = useCallback(
info => {
- if (info.file.status === 'uploading') {
+ const {status, response} = info.file;
+
+ if (status === 'uploading') {
onUploadChange(LOADING);
+ return;
+ }
+
+ // antd emits 'error' on HTTP/network failure. Without this branch the
+ // canvas stays in LOADING forever. An error page body is not JSON, so
+ // `response.message` may be missing entirely.
+ if (status === 'error') {
+ const failure = response?.message || t('analysis.canvas.import_failed');
+ onUploadChange(UPLOAD_FAILED, failure);
+ message.error(failure);
+ return;
}
- if (info.file.status === 'done') {
- const {response} = info.fileList[0];
- if (response.status === 200) {
+
+ if (status === 'done') {
+ if (response?.status === 200) {
onUploadChange(SUCCESS, t('analysis.canvas.import_success'), response.data);
message.success(t('analysis.canvas.import_success'));
return;
}
- onUploadChange(UPLOAD_FAILED, response.message);
+
+ const failure = response?.message || t('analysis.canvas.import_failed');
+ onUploadChange(UPLOAD_FAILED, failure);
}
},
[onUploadChange, t]
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.js b/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.js
index 629053845..a38f23f53 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.js
@@ -18,7 +18,7 @@
import {Modal, Button, Form, Input, Typography, Select,
Divider, Upload, Radio, AutoComplete, Checkbox, message, Space} from 'antd';
-import {useState, useCallback} from 'react';
+import {useState, useCallback, useEffect, useRef} from 'react';
import {useTranslation} from 'react-i18next';
import * as api from '../../api';
import * as rules from '../../utils/rules';
@@ -937,6 +937,20 @@ const EditLayer = ({edit, visible, onCancel, refresh}) => {
const [selectedTemplate, setSelectedTemplate] = useState(undefined);
const [loading, setLoading] = useState(false);
const [form] = Form.useForm();
+ const requestEpoch = useRef(0);
+ const visibleRef = useRef(visible);
+ visibleRef.current = visible;
+
+ const invalidateRequests = useCallback(() => {
+ requestEpoch.current += 1;
+ setLoading(false);
+ }, []);
+
+ useEffect(() => {
+ if (!visible) {
+ invalidateRequests();
+ }
+ }, [invalidateRequests, visible]);
const sourceTypeOptions = [
{label: 'HDFS', value: 'HDFS'},
@@ -950,30 +964,59 @@ const EditLayer = ({edit, visible, onCancel, refresh}) => {
return;
}
+ const session = requestEpoch.current;
form.validateFields().then(values => {
+ if (!visibleRef.current || session !== requestEpoch.current) {
+ return;
+ }
const formatData = formatDatasource(values);
- // return;
+ const request = ++requestEpoch.current;
setLoading(true);
+ // Every active exit clears `loading`; callbacks from an invalidated
+ // modal session are ignored so they cannot affect a reopened form.
api.manage.addDatasource(formatData).then(res => {
+ if (!visibleRef.current || request !== requestEpoch.current) {
+ return;
+ }
setLoading(false);
- if (res.status === 200) {
+ if (res?.status === 200) {
message.success(t('common.msg.create_success'));
+ requestEpoch.current += 1;
onCancel();
refresh();
return;
}
- message.error(res.message);
+ message.error(res?.message || t('common.msg.operation_failed'));
+ }).catch(() => {
+ if (!visibleRef.current || request !== requestEpoch.current) {
+ return;
+ }
+ setLoading(false);
+ message.error(t('common.msg.operation_failed'));
});
+ }).catch(error => {
+ if (!visibleRef.current || session !== requestEpoch.current) {
+ return;
+ }
+ if (!error?.errorFields) {
+ message.error(t('common.msg.operation_failed'));
+ }
});
}, [form, loading, onCancel, refresh, t]);
+ const handleCancel = useCallback(() => {
+ invalidateRequests();
+ onCancel();
+ }, [invalidateRequests, onCancel]);
+
const handleClose = useCallback(() => {
+ invalidateRequests();
form.resetFields();
setSourceType('');
setSelectedTemplate(undefined);
- }, [form]);
+ }, [form, invalidateRequests]);
const handleType = useCallback(val => {
form.resetFields(SOURCE_CONFIG_FIELDS);
@@ -989,7 +1032,7 @@ const EditLayer = ({edit, visible, onCancel, refresh}) => {
return (
<Modal
title={edit ? t('datasource.form.title_edit') : t('datasource.form.title_create')}
- onCancel={onCancel}
+ onCancel={handleCancel}
open={visible}
width={600}
onOk={onFinish}
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.test.js b/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.test.js
index b5cae6c0e..4ce5fef33 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.test.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Datasource/EditLayer.test.js
@@ -16,7 +16,7 @@
*/
import {useCallback, useState} from 'react';
-import {fireEvent, render, screen, waitFor} from '@testing-library/react';
+import {act, fireEvent, render, screen, waitFor} from '@testing-library/react';
import userEvent from '@testing-library/user-event';
import EditLayer from './EditLayer';
@@ -52,6 +52,7 @@ const ReopenableEditLayer = ({refresh = jest.fn()}) => {
return (
<>
<button onClick={open}>reopen datasource</button>
+ <button onClick={close}>close datasource externally</button>
<EditLayer
visible={visible}
edit={false}
@@ -68,6 +69,16 @@ const selectTemplate = async label => {
await userEvent.click(await screen.findByText(label));
};
+const deferred = () => {
+ let resolve;
+ let reject;
+ const promise = new Promise((onResolve, onReject) => {
+ resolve = onResolve;
+ reject = onReject;
+ });
+ return {promise, resolve, reject};
+};
+
const expectFreshForm = () => {
expect(screen.getByPlaceholderText('datasource.form.name_placeholder')).toHaveValue('');
expect(screen.queryByText('FILE', {selector: '.ant-select-selection-item'}))
@@ -144,3 +155,73 @@ test('clears template, type, and hidden config after successful create and reope
await userEvent.click(screen.getByRole('button', {name: 'reopen datasource'}));
expectFreshForm();
});
+
+test('ignores a create response from a closed modal session', async () => {
+ const api = jest.requireMock('../../api');
+ const first = deferred();
+ const second = deferred();
+ const refresh = jest.fn();
+ api.manage.addDatasource.mockReset();
+ api.manage.addDatasource
+ .mockReturnValueOnce(first.promise)
+ .mockReturnValueOnce(second.promise);
+ render(<ReopenableEditLayer refresh={refresh} />);
+
+ await selectTemplate('datasource.form.templates.jdbc_mysql');
+ fireEvent.change(screen.getByPlaceholderText('datasource.form.url_placeholder'), {
+ target: {value: 'jdbc:mysql://127.0.0.1:3306/first'},
+ });
+ fireEvent.change(screen.getByPlaceholderText('datasource.form.password_placeholder'), {
+ target: {value: 'first-secret'},
+ });
+ await userEvent.click(screen.getByRole('button', {name: 'OK'}));
+ await waitFor(() => expect(api.manage.addDatasource).toHaveBeenCalledTimes(1));
+
+ await userEvent.click(screen.getByRole('button', {name: 'Cancel'}));
+ await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument());
+ await userEvent.click(screen.getByRole('button', {name: 'reopen datasource'}));
+ await selectTemplate('datasource.form.templates.jdbc_mysql');
+ fireEvent.change(screen.getByPlaceholderText('datasource.form.url_placeholder'), {
+ target: {value: 'jdbc:mysql://127.0.0.1:3306/second'},
+ });
+ fireEvent.change(screen.getByPlaceholderText('datasource.form.password_placeholder'), {
+ target: {value: 'second-secret'},
+ });
+ await userEvent.click(screen.getByRole('button', {name: 'OK'}));
+ await waitFor(() => expect(api.manage.addDatasource).toHaveBeenCalledTimes(2));
+
+ await act(async () => first.resolve({status: 200}));
+ expect(screen.getByRole('dialog')).toBeInTheDocument();
+ expect(refresh).not.toHaveBeenCalled();
+
+ await act(async () => second.resolve({status: 200}));
+ await waitFor(() => expect(screen.queryByRole('dialog')).not.toBeInTheDocument());
+ expect(refresh).toHaveBeenCalledTimes(1);
+});
+
+test('ignores a create response after prop-driven modal closure', async () => {
+ const api = jest.requireMock('../../api');
+ const request = deferred();
+ const refresh = jest.fn();
+ api.manage.addDatasource.mockReset();
+ api.manage.addDatasource.mockReturnValueOnce(request.promise);
+ render(<ReopenableEditLayer refresh={refresh} />);
+
+ await selectTemplate('datasource.form.templates.jdbc_mysql');
+ fireEvent.change(screen.getByPlaceholderText('datasource.form.url_placeholder'), {
+ target: {value: 'jdbc:mysql://127.0.0.1:3306/closed'},
+ });
+ fireEvent.change(screen.getByPlaceholderText('datasource.form.password_placeholder'), {
+ target: {value: 'temporary-secret'},
+ });
+ await userEvent.click(screen.getByRole('button', {name: 'OK'}));
+ await waitFor(() => expect(api.manage.addDatasource).toHaveBeenCalledTimes(1));
+
+ await userEvent.click(screen.getByRole('button', {
+ name: 'close datasource externally',
+ }));
+ await act(async () => request.resolve({status: 200}));
+
+ expect(refresh).not.toHaveBeenCalled();
+ expect(screen.queryByRole('dialog')).not.toBeInTheDocument();
+});
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Graph/index.js b/hugegraph-hubble/hubble-fe/src/pages/Graph/index.js
index 4c8fcb70a..a07696bfa 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Graph/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Graph/index.js
@@ -351,14 +351,14 @@ const Graph = () => {
dataIndex: 'create_time',
align: 'center',
width: 140,
- render: val => moment(val).format('YYYY-MM-DD'),
+ render: val => (val ? moment(val).format('YYYY-MM-DD') : '--'),
},
{
title: t('graph.detail.update_data'),
dataIndex: 'update_time',
align: 'center',
width: 140,
- render: val => moment(val).format('YYYY-MM-DD'),
+ render: val => (val ? moment(val).format('YYYY-MM-DD') : '--'),
},
{
title: t('graph.col.creator'),
@@ -615,6 +615,7 @@ const Graph = () => {
columns={columns}
dataSource={data}
rowKey='name'
+ scroll={{x: 'max-content'}}
locale={{emptyText: listUnavailable ? null : emptyState}}
pagination={pagination}
onChange={handleTable}
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Graph/index.module.scss b/hugegraph-hubble/hubble-fe/src/pages/Graph/index.module.scss
index 455a63f48..6c5c6df8f 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Graph/index.module.scss
+++ b/hugegraph-hubble/hubble-fe/src/pages/Graph/index.module.scss
@@ -70,7 +70,11 @@ div.add_card {
.element_counts {
display: inline-block;
+ max-width: 100%;
+ overflow: hidden;
white-space: nowrap;
+ text-overflow: ellipsis;
+ vertical-align: middle;
color: #667085;
cursor: default;
}
diff --git a/hugegraph-hubble/hubble-fe/src/pages/GraphSpace/Card.js b/hugegraph-hubble/hubble-fe/src/pages/GraphSpace/Card.js
index 6b369231e..f4db39cd6 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/GraphSpace/Card.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/GraphSpace/Card.js
@@ -55,7 +55,7 @@ const TitleField = ({item, onClick, onKeyDown}) => {
{item.default && (
<span className={style.default}>{t('graphspace.card.set_default')}</span>
)}
- {moment(item.create_time).format('YYYY-MM-DD')} {t('graphspace.card.created')}
+ {item.create_time ? moment(item.create_time).format('YYYY-MM-DD') : '--'} {t('graphspace.card.created')}
</div>
</>
);
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Meta/Edge/index.js b/hugegraph-hubble/hubble-fe/src/pages/Meta/Edge/index.js
index e4749510f..873817c85 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Meta/Edge/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Meta/Edge/index.js
@@ -169,10 +169,10 @@ const EdgeTable = () => {
];
useEffect(() => {
- api.manage.getMetaPropertyList(graphspace, graph).then(res => {
+ api.manage.getMetaPropertyList(graphspace, graph, {page_size: -1}).then(res => {
if (res.status === 200) {
setPropertyList(res.data.records.map(item => ({
- lable: item.name,
+ label: item.name,
value: item.name,
data_type: item.data_type,
})));
@@ -215,6 +215,7 @@ const EdgeTable = () => {
rowKey={rowKey}
showExpandColumn={false}
loading={loading}
+ scroll={{x: 'max-content'}}
/>
<EditEdgeLayer
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Meta/Vertex/index.js b/hugegraph-hubble/hubble-fe/src/pages/Meta/Vertex/index.js
index fe588d5ec..e7fd924ec 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Meta/Vertex/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Meta/Vertex/index.js
@@ -172,7 +172,7 @@ const VertexTable = () => {
api.manage.getMetaPropertyList(graphspace, graph, {page_size: -1}).then(res => {
if (res.status === 200) {
setPropertyList(res.data.records.map(item => ({
- lable: item.name,
+ label: item.name,
value: item.name,
data_type: item.data_type,
})));
@@ -204,6 +204,7 @@ const VertexTable = () => {
onChange={handleTable}
rowKey={rowKey}
loading={loading}
+ scroll={{x: 'max-content'}}
/>
<EditVertexLayer
diff --git a/hugegraph-hubble/hubble-fe/src/pages/My/index.js b/hugegraph-hubble/hubble-fe/src/pages/My/index.js
index 90e7b26bc..e3ac4896a 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/My/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/My/index.js
@@ -244,7 +244,11 @@ const My = () => {
<Form.Item label={t('my.col.name')} name='user_name'>
<Input disabled />
</Form.Item>
- <Form.Item label={t('my.edit.old_password')} name='old_password'>
+ <Form.Item
+ label={t('my.edit.old_password')}
+ name='old_password'
+ rules={[rules.required()]}
+ >
<Input.Password
autoComplete='new-password'
placeholder={t('my.edit.old_password_placeholder')}
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Operations/Nodes.js b/hugegraph-hubble/hubble-fe/src/pages/Operations/Nodes.js
index 51856efa1..a9b35ee9d 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Operations/Nodes.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Operations/Nodes.js
@@ -192,7 +192,9 @@ const Nodes = () => {
const changeTable = useCallback((pagination, filters, sorter) => update({
page: String(pagination.current),
page_size: String(pagination.pageSize),
- sort: sorter.field || 'name',
+ // Columns without a dataIndex (observed_at) leave sorter.field undefined,
+ // so fall back to the column key that sortOrder() also matches on.
+ sort: (sorter.columnKey ?? sorter.field) || 'name',
order: sorter.order === 'descend' ? 'desc' : 'asc',
}), [update]);
const sortOrder = field => {
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Task/components/EditLayer.js b/hugegraph-hubble/hubble-fe/src/pages/Task/components/EditLayer.js
index ebb37984f..3886841d8 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Task/components/EditLayer.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Task/components/EditLayer.js
@@ -23,7 +23,7 @@ import {
Radio,
message,
} from 'antd';
-import {useEffect, useState, useCallback} from 'react';
+import {useEffect, useState, useCallback, useRef} from 'react';
import {useTranslation} from 'react-i18next';
import * as rules from '../../../utils/rules';
import * as api from '../../../api';
@@ -32,6 +32,8 @@ const EditLayer = ({visible, onCancel, data, refresh}) => {
const {t} = useTranslation();
const [form] = Form.useForm();
const [syncType, setSyncType] = useState('');
+ const [loading, setLoading] = useState(false);
+ const updateRequest = useRef(null);
const datasourceType = data?.ingestion_mapping?.structs[0]?.input?.type;
const scheduleOptions = datasourceType === 'KAFKA'
? [{label: t('task.sync.realtime'), value: 'REALTIME'}]
@@ -51,20 +53,44 @@ const EditLayer = ({visible, onCancel, data, refresh}) => {
delete values.task_schedule_extend;
}
+ const request = Symbol('task-update');
+ updateRequest.current = request;
+ setLoading(true);
+ // The OK spinner must clear on every outcome, including a rejected
+ // request, otherwise the button stays locked with the modal open.
api.manage.updateTask(data.task_id, values).then(res => {
- if (res.status === 200) {
+ if (updateRequest.current !== request) {
+ return;
+ }
+ updateRequest.current = null;
+ setLoading(false);
+
+ if (res?.status === 200) {
message.success(t('task.edit.update_success'));
onCancel();
refresh();
return;
}
- message.error(res.message);
+ message.error(res?.message || t('common.msg.operation_failed'));
+ }).catch(() => {
+ if (updateRequest.current !== request) {
+ return;
+ }
+ updateRequest.current = null;
+ setLoading(false);
+ message.error(t('common.msg.operation_failed'));
});
+ }).catch(error => {
+ if (!error?.errorFields) {
+ message.error(t('common.msg.operation_failed'));
+ }
});
}, [data.task_id, form, onCancel, refresh, syncType, t]);
useEffect(() => {
+ updateRequest.current = null;
+ setLoading(false);
if (!visible) {
return;
}
@@ -78,14 +104,24 @@ const EditLayer = ({visible, onCancel, data, refresh}) => {
message.error(res.message);
});
+ return () => {
+ updateRequest.current = null;
+ };
}, [visible, data.task_id, form]);
+ const handleCancel = useCallback(() => {
+ updateRequest.current = null;
+ setLoading(false);
+ onCancel();
+ }, [onCancel]);
+
return (
<Modal
title={t('task.edit_title_edit')}
- onCancel={onCancel}
+ onCancel={handleCancel}
open={visible}
onOk={onFinish}
+ confirmLoading={loading}
destroyOnClose
>
<Form
diff --git a/hugegraph-hubble/hubble-fe/src/pages/Task/index.js b/hugegraph-hubble/hubble-fe/src/pages/Task/index.js
index f1ffe0bc7..db43435cc 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/Task/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/Task/index.js
@@ -260,13 +260,17 @@ const Task = () => {
setLoading(true);
api.manage.deleteTask(id).then(res => {
setLoading(false);
- if (res.status === 200) {
+
+ if (res?.status === 200) {
setRefresh(!refresh);
message.success(t('common.msg.delete_success'));
return;
}
+ message.error(t('common.msg.delete_fail'));
+ }).catch(() => {
+ setLoading(false);
message.error(t('common.msg.delete_fail'));
});
}, [refresh, t]);
@@ -382,19 +386,19 @@ const Task = () => {
{
title: t('task.col.target_space'),
dataIndex: 'ingestion_option',
- render: val => val.graphspace,
+ render: val => val?.graphspace ?? '-',
},
{
title: t('task.col.target_graph'),
dataIndex: 'ingestion_option',
- render: val => val.graph,
+ render: val => val?.graph ?? '-',
},
{
title: t('task.col.create_time'),
dataIndex: 'create_time',
},
{
- title: t('account.col.id'),
+ title: t('task.col.creator'),
dataIndex: 'creator',
},
{
@@ -617,6 +621,7 @@ const Task = () => {
<Table
columns={columns}
rowKey={rowKey}
+ scroll={{x: 'max-content'}}
dataSource={data}
pagination={pagination}
onChange={handleTable}
diff --git a/hugegraph-hubble/hubble-fe/src/pages/TaskDetail/index.js b/hugegraph-hubble/hubble-fe/src/pages/TaskDetail/index.js
index 39ca3865e..1bf23d70f 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/TaskDetail/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/TaskDetail/index.js
@@ -28,7 +28,7 @@ const createColumns = t => [
{
title: t('task.detail.job_id'),
dataIndex: 'job_id',
- render: val => val.toString(),
+ render: val => (val === null || val === undefined ? '-' : val.toString()),
},
{
title: t('task.detail.import_count'),
@@ -81,6 +81,8 @@ const createColumns = t => [
},
];
+const getRowKey = record => record.id ?? record.job_id;
+
const TaskDetail = () => {
const {t} = useTranslation();
const columns = createColumns(t);
@@ -161,9 +163,10 @@ const TaskDetail = () => {
)}
<Spin spinning={loading}>
<Table
- rowKey='job_id'
+ rowKey={getRowKey}
columns={columns}
dataSource={visibleData}
+ scroll={{x: 'max-content'}}
/>
</Spin>
</div>
diff --git a/hugegraph-hubble/hubble-fe/src/pages/TaskEdit/BaseForm/index.js b/hugegraph-hubble/hubble-fe/src/pages/TaskEdit/BaseForm/index.js
index 0277f9c93..c1357c3db 100644
--- a/hugegraph-hubble/hubble-fe/src/pages/TaskEdit/BaseForm/index.js
+++ b/hugegraph-hubble/hubble-fe/src/pages/TaskEdit/BaseForm/index.js
@@ -84,12 +84,21 @@ const BaseForm = ({cancel, visible, loading}) => {
const checkExistName = () => ({
validator: async (_, value) => {
- const res = await api.manage.getTaskList({query: value}).then(res => {
+ const res = await api.manage.getTaskList({
+ query: value,
+ // `query` is a substring match, so the exact name can fall outside
+ // the default first page of 10 substring hits.
+ page_size: -1,
+ }).then(res => {
if (res.status !== 200) {
return t('task.edit.duplicate_check_failed');
}
- if (res.data.total > 0) {
+ // The list endpoint treats `query` as a LIKE '%value%' match, so a
+ // non-zero total only means some name contains this one. Compare the
+ // returned names exactly before rejecting.
+ const records = res.data?.records ?? [];
+ if (records.some(item => item.task_name === value)) {
return t('task.edit.duplicate_name');
}
diff --git a/hugegraph-hubble/hubble-fe/src/utils/formatTimeDuration.js b/hugegraph-hubble/hubble-fe/src/utils/formatTimeDuration.js
index bcfdfc00c..6fe7a52ab 100644
--- a/hugegraph-hubble/hubble-fe/src/utils/formatTimeDuration.js
+++ b/hugegraph-hubble/hubble-fe/src/utils/formatTimeDuration.js
@@ -27,10 +27,16 @@ import _ from 'lodash';
export default function formatTimeDuration(start, end) {
const duration = intervalToDuration({start, end});
- const {days, hours, minutes, seconds} = duration || {};
- const preFormatedDuration = formatDuration(duration, {format: ['days', 'hours', 'minutes', 'seconds']});
+ const preFormatedDuration = formatDuration(
+ duration,
+ {format: ['years', 'months', 'days', 'hours', 'minutes', 'seconds']}
+ );
const replaceMap = {
+ ' years': 'y',
+ ' year': 'y',
+ ' months': 'mo',
+ ' month': 'mo',
' days': 'd',
' day': 'd',
' hours': 'h',
@@ -47,8 +53,7 @@ export default function formatTimeDuration(start, end) {
formatedDuration = _.replace(formatedDuration, key, replaceMap[key]);
}
);
- const restMilliSecond = end - start - seconds * 1000 - minutes * 1000 * 60
- - hours * 1000 * 60 * 60 - days * 1000 * 60 * 60 * 24;
+ const restMilliSecond = (end - start) % 1000;
formatedDuration = formatedDuration + ` ${restMilliSecond}ms`;
return formatedDuration;