Apache TinkerPop•3w ago

Unable to delete a vertex in AWS Neptune even after removing all its 50k+ edges

Hi everyone, I’m running into a strange issue with AWS Neptune and would appreciate any insights. I’m trying to delete a vertex that used to have over 50,000 edges. Before attempting the deletion, I made sure to remove all the relationships connected to it (using both bothE().drop() and validations to confirm there are no remaining edges). When I validate the vertex afterward, it shows 0 relationships, but when I try to delete it with something like: g.V(vertexId).drop().iterate() the operation just hangs and never completes — almost as if the vertex were still “locked” or referencing something internally. Has anyone seen this behavior before on Neptune? Could it be some kind of internal index or cleanup process that prevents deleting large, previously connected vertices? Any advice or explanation about this behavior would be really helpful. Thanks in advance! — Hugo

21 Replies

spmallette•3w ago

In what environment are you doing this? For example, are you executing these deletions from graph-notebook? using the aws sdk? something else?

masterhugoOP•3w ago

I’m executing this query using gremlin python on a aws Lambda, with query timeout of 300 seconds, all good connection with vpc and aws Neptune, even I know I can delete nodes with less than 10k edges with no problem, but with more edges, colapse the database and get timeout

spmallette•3w ago

That sounds a little different than your original question. I hear now you are experiencing timeouts. are those on the deleting of the edges or the single remaining verex?

masterhugoOP•3w ago

I’m getting a timeout when trying to delete a single vertex by its ID, even though I supposedly deleted all its edges beforehand. I also noticed that when deleting the vertex that previously had around 10k relationships, the operation eventually succeeded — but it took a long time, even though I had supposedly deleted all its edges beforehand.

spmallette•3w ago

you said that you are using gremlinpython for this. are you using a session to do this? or doing this in a transaction with g.tx()?

masterhugoOP•3w ago

I’m using g.tx()

spmallette•3w ago

Is the entire transaction designed to just delete one vertex (and its' associated edges)? or is there more happening?

masterhugoOP•3w ago

In this particular case, is just one transaction to delete one vertex, not the edges, that happen in another transaction in another lambda, outside the environment and it happen a long time ago before the delete operation of the vertex

spmallette•3w ago

oh...hmm, that's interesting and, in between the deletion of edges that happened a long time ago and deletion request for the vertex you were able to confirm by separate request that the vertex had zero edges?

masterhugoOP•3w ago

That’s right, I made a validation before to know it have 0 edges in both_e, in_e, out_e, with limit(1) and count, it always said 0

spmallette•3w ago

sorry for trying to get really specific here. just to be clear, this validation of edge deletion was done after the transaction for deleting edges was committed with g.tx().commit()?

masterhugoOP•3w ago

That’s right No problem, I’m really confused with this, cause I don’t understand what is happening, the theory it should be doing that but in reality is something weird

Andrea•3w ago

maybe posting the code could troubleshooting? could something about the vertex other than its edges be affecting it? perhaps 50 000 properties?

masterhugoOP•3w ago

Only have 10 properties, and here is the code

class GremlinTransaction:
def __init__(self, remote: graph_traversal.GraphTraversalSource) -> None:
self.g_transaction = remote

def __enter__(self) -> graph_traversal.GraphTraversalSource:
self.tx: graph_traversal.Transaction = self.g_transaction.tx()
return self.tx.begin()

def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
del exc_type, traceback
if self.tx.is_open():
if isinstance(exc_value, Exception):
self.tx.rollback()
return

try:
self.tx.commit()
except Exception:
self.tx.rollback()
def delete_node_by_id(self, node_id: str) -> None:
_g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
with self.GremlinTransaction(_g) as transaction:
g_traversal = transaction.get_graph_traversal()
g_traversal.V(node_id).drop().iterate() 
class DeleteNodesExecutionHandler(AbstractGraphExecutionHandler):
def __init__(
self,
cmd: commands.BaseGraphCommand,
graph_write_repository: graph_adapter.GraphWriteRepository,
) -> None:
super().__init__(
cmd=cmd,
graph_write_repository=graph_write_repository,
)

def execute(self) -> None:
_LOGGER.info(f"Starting node deletion for nodes with label: {NODE_LABEL}")
if self.write_repo is None:
raise RuntimeError(RUNTIME_ERROR_MESSAGE)
for node in NODES_LIST:
try:
_LOGGER.info("start to delete node %s", node)
self.write_repo.delete_node_by_id(node_id=node)
_LOGGER.info("Node %s deleted successfully", node)
except graph_exceptions.TimeLimitExceededException:
_LOGGER.error("Node deleted error")
return None

class GremlinTransaction:
def __init__(self, remote: graph_traversal.GraphTraversalSource) -> None:
self.g_transaction = remote

def __enter__(self) -> graph_traversal.GraphTraversalSource:
self.tx: graph_traversal.Transaction = self.g_transaction.tx()
return self.tx.begin()

def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
del exc_type, traceback
if self.tx.is_open():
if isinstance(exc_value, Exception):
self.tx.rollback()
return

try:
self.tx.commit()
except Exception:
self.tx.rollback()
def delete_node_by_id(self, node_id: str) -> None:
_g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
with self.GremlinTransaction(_g) as transaction:
g_traversal = transaction.get_graph_traversal()
g_traversal.V(node_id).drop().iterate() 
class DeleteNodesExecutionHandler(AbstractGraphExecutionHandler):
def __init__(
self,
cmd: commands.BaseGraphCommand,
graph_write_repository: graph_adapter.GraphWriteRepository,
) -> None:
super().__init__(
cmd=cmd,
graph_write_repository=graph_write_repository,
)

def execute(self) -> None:
_LOGGER.info(f"Starting node deletion for nodes with label: {NODE_LABEL}")
if self.write_repo is None:
raise RuntimeError(RUNTIME_ERROR_MESSAGE)
for node in NODES_LIST:
try:
_LOGGER.info("start to delete node %s", node)
self.write_repo.delete_node_by_id(node_id=node)
_LOGGER.info("Node %s deleted successfully", node)
except graph_exceptions.TimeLimitExceededException:
_LOGGER.error("Node deleted error")
return None

Andrea•3w ago

sorry I might be sort of blind but I don't see any code which is deleting edges specifically before the node itself?

masterhugoOP•2w ago

Sry, here is the other parts

def count_edges_by_node(self, node: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node).both_e().count().next()
return query

def has_edges_by_node(self, node: str, limit: int = 1) -> list[str]:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node).both_e().id_().limit(limit).to_list()
return query

def has_nodes_by_label(self, label: str, limit: int = 1) -> list[str]:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V().has_label(label).id_().limit(limit).to_list()
return query

def validation_node_edges_both_e(self, node_id: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node_id).both_e().limit(1).count().next()
return query

def validation_node_edges_in(self, node_id: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node_id).in_e().limit(1).count().next()
return query

def validation_node_edges_out(self, node_id: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node_id).out_e().limit(1).count().next()
return query

def count_edges_by_node(self, node: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node).both_e().count().next()
return query

def has_edges_by_node(self, node: str, limit: int = 1) -> list[str]:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node).both_e().id_().limit(limit).to_list()
return query

def has_nodes_by_label(self, label: str, limit: int = 1) -> list[str]:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V().has_label(label).id_().limit(limit).to_list()
return query

def validation_node_edges_both_e(self, node_id: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node_id).both_e().limit(1).count().next()
return query

def validation_node_edges_in(self, node_id: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node_id).in_e().limit(1).count().next()
return query

def validation_node_edges_out(self, node_id: str) -> int:
g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
query = g.V(node_id).out_e().limit(1).count().next()
return query

class ValidationEdgesExecutionHandler(AbstractGraphExecutionHandler):
def __init__(
self,
cmd: commands.BaseGraphCommand,
graph_write_repository: graph_adapter.GraphWriteRepository,
) -> None:
super().__init__(
cmd=cmd,
graph_write_repository=graph_write_repository,
)

def execute(self) -> None:
_LOGGER.info(f"Starting node deletion for nodes with label: {NODE_LABEL}")
if self.write_repo is None:
raise RuntimeError(RUNTIME_ERROR_MESSAGE)
for node in NODES_LIST:
try:
_LOGGER.info("start to validate node %s", node)
validation = self.write_repo.validation_node_edges_both_e(node_id=node)
_LOGGER.info("both validation %d", validation)
validation = self.write_repo.validation_node_edges_in(node_id=node)
_LOGGER.info("in validation %d", validation)
validation = self.write_repo.validation_node_edges_out(node_id=node)
_LOGGER.info("out validation %d", validation)
_LOGGER.info("End validation node %s successfully", node)
except graph_exceptions.TimeLimitExceededException:
_LOGGER.error("Node deleted error")
return None

class ValidationEdgesExecutionHandler(AbstractGraphExecutionHandler):
def __init__(
self,
cmd: commands.BaseGraphCommand,
graph_write_repository: graph_adapter.GraphWriteRepository,
) -> None:
super().__init__(
cmd=cmd,
graph_write_repository=graph_write_repository,
)

def execute(self) -> None:
_LOGGER.info(f"Starting node deletion for nodes with label: {NODE_LABEL}")
if self.write_repo is None:
raise RuntimeError(RUNTIME_ERROR_MESSAGE)
for node in NODES_LIST:
try:
_LOGGER.info("start to validate node %s", node)
validation = self.write_repo.validation_node_edges_both_e(node_id=node)
_LOGGER.info("both validation %d", validation)
validation = self.write_repo.validation_node_edges_in(node_id=node)
_LOGGER.info("in validation %d", validation)
validation = self.write_repo.validation_node_edges_out(node_id=node)
_LOGGER.info("out validation %d", validation)
_LOGGER.info("End validation node %s successfully", node)
except graph_exceptions.TimeLimitExceededException:
_LOGGER.error("Node deleted error")
return None

def delete_edges_by_label(self, edge_ids: list[str]) -> None:
_g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
with self.GremlinTransaction(_g) as transaction:
g_traversal = transaction.get_graph_traversal()
g_traversal.E(*edge_ids).drop().iterate() 
class DeleteGraphExecutionHandler(AbstractGraphExecutionHandler):
def __init__(
self,
cmd: commands.BaseGraphCommand,
graph_write_repository: graph_adapter.GraphWriteRepository,
) -> None:
super().__init__(
cmd=cmd,
graph_write_repository=graph_write_repository,
)

def execute(self) -> None:
_LOGGER.info(f"Starting edge deletion for nodes with label: {NODE_LABEL}")
if self.write_repo is None:
raise RuntimeError(RUNTIME_ERROR_MESSAGE)
_LOGGER.info("Gather nodes to delete")

id_nodes = self.write_repo.has_nodes_by_label(
label=NODE_LABEL, limit=DELETE_BATCH_SIZE
)
iteration_number = 0
for node in id_nodes:
try:
total_edges = self.write_repo.count_edges_by_node(node=node)
_LOGGER.info("total edges on the node %s: %d", node, total_edges)
while iteration_number < MAX_ITERATIONS and total_edges > 0:
_LOGGER.info("start to delete edges")
id_edges = self.write_repo.has_edges_by_node(
node=node, limit=DELETE_BATCH_SIZE
)
if len(id_edges) == 0:
_LOGGER.info(
"No more edges found. Edge deletion phase complete."
)
break

self.write_repo.delete_edges_by_label(edge_ids=id_edges)
_LOGGER.info("Delete edges %d", len(id_edges))

time.sleep(SLEEP_SECONDS)
iteration_number += 1
if iteration_number >= MAX_ITERATIONS:
_LOGGER.warning(
f"Edge deletion stopped after reaching MAX_ITERATIONS ({MAX_ITERATIONS}). "
"Some edges might still remain."
)
return None
except graph_exceptions.TimeLimitExceededException:
_LOGGER.error("Trying to find the error")

time.sleep(SLEEP_SECONDS)
return None

def delete_edges_by_label(self, edge_ids: list[str]) -> None:
_g = self._get_traversal(timeout=constants.DEFAULT_QUERY_DELETE_TIMEOUT)
with self.GremlinTransaction(_g) as transaction:
g_traversal = transaction.get_graph_traversal()
g_traversal.E(*edge_ids).drop().iterate() 
class DeleteGraphExecutionHandler(AbstractGraphExecutionHandler):
def __init__(
self,
cmd: commands.BaseGraphCommand,
graph_write_repository: graph_adapter.GraphWriteRepository,
) -> None:
super().__init__(
cmd=cmd,
graph_write_repository=graph_write_repository,
)

def execute(self) -> None:
_LOGGER.info(f"Starting edge deletion for nodes with label: {NODE_LABEL}")
if self.write_repo is None:
raise RuntimeError(RUNTIME_ERROR_MESSAGE)
_LOGGER.info("Gather nodes to delete")

id_nodes = self.write_repo.has_nodes_by_label(
label=NODE_LABEL, limit=DELETE_BATCH_SIZE
)
iteration_number = 0
for node in id_nodes:
try:
total_edges = self.write_repo.count_edges_by_node(node=node)
_LOGGER.info("total edges on the node %s: %d", node, total_edges)
while iteration_number < MAX_ITERATIONS and total_edges > 0:
_LOGGER.info("start to delete edges")
id_edges = self.write_repo.has_edges_by_node(
node=node, limit=DELETE_BATCH_SIZE
)
if len(id_edges) == 0:
_LOGGER.info(
"No more edges found. Edge deletion phase complete."
)
break

self.write_repo.delete_edges_by_label(edge_ids=id_edges)
_LOGGER.info("Delete edges %d", len(id_edges))

time.sleep(SLEEP_SECONDS)
iteration_number += 1
if iteration_number >= MAX_ITERATIONS:
_LOGGER.warning(
f"Edge deletion stopped after reaching MAX_ITERATIONS ({MAX_ITERATIONS}). "
"Some edges might still remain."
)
return None
except graph_exceptions.TimeLimitExceededException:
_LOGGER.error("Trying to find the error")

time.sleep(SLEEP_SECONDS)
return None

Andrea•2w ago

if you add a log in the same transaction before you delete the node to output the number of edges, can you verify that it has zero edges? i am wondering if there's an issue with the transaction boundaries of the transaction deleting the edges vs the one deleting the node

masterhugoOP•2w ago

Yep, I did that and have the same value

Andrea•2w ago

Is it possible to try performing edge deletion and node deletion in a single transaction?

spmallette•2w ago

not sure what options you have open to you, but this might be a case that needs more official support from the Neptune team (i.e. a support ticket).

masterhugoOP•2w ago

Got it, thank you for the suggestion! I’ll go ahead and open a support ticket with the Neptune team to look into this further. Really appreciate your help! 🥹

Gaming

Programming

Unable to delete a vertex in AWS Neptune even after removing all its 50k+ edges

Did you find this page helpful?