[Core][DAG] Deprecate DAGNode.execute()

[Phucvt123]

Why are these changes needed?

In the non-compiled DAG path, calling DAGNode.execute() executes FunctionNode.execute(), which dynamically defines a new remote function via ray.remote(self._body) on every execution. This exports new metadata to the GCS KV store on every run, leading to an unbounded memory leak (GCS KV leak) during the job lifetime (#63666).

Since the non-compiled DAG execution path is not recommended for production and is a primary source of this leak, we are deprecating DAGNode.execute() in favor of the compiled DAG API.

This PR adds a DeprecationWarning to DAGNode.execute() to warn users that it is deprecated and will be removed in a future release.

Related issue number

Closes #63666

[gemini-code-assist]

Code Review

This pull request introduces a class-level cache (_remote_func_cache) in FunctionNode to prevent re-registering the same function in GCS KV on every execution, along with corresponding unit tests. The reviewer pointed out that using a class-level dictionary to store RemoteFunction objects creates a permanent memory leak because it holds strong references to the function bodies, preventing garbage collection. The reviewer suggested storing the cached RemoteFunction directly as an attribute on the function body itself (with a fallback to the class-level cache for objects that do not support setting attributes) and provided corresponding updates for the unit tests.

[gemini-code-assist]

Using a class-level dictionary (_remote_func_cache) to store RemoteFunction objects creates a permanent memory leak.

Because RemoteFunction holds a strong reference to the underlying function body (self._body), and the class-level cache holds a strong reference to the RemoteFunction, any function executed via a DAG will never be garbage collected. In long-running applications that dynamically define functions or use closures, this will lead to unbounded memory growth.

To fix this, we can attempt to store the cached RemoteFunction directly as an attribute on the function body itself. This ties the lifetime of the cached RemoteFunction to the function body. For objects that do not support setting attributes (like built-ins), we can fall back to the class-level cache.

Suggested change

	func_id = id(self._body)
	if func_id not in FunctionNode._remote_func_cache:
	FunctionNode._remote_func_cache[func_id] = ray.remote(self._body)
	return (
	ray.remote(self._body)
	FunctionNode._remote_func_cache[func_id]
	.options(**self._bound_options)
	.remote(*self._bound_args, **self._bound_kwargs)
	)
	func = self._body
	try:
	if not hasattr(func, "_ray_remote_func"):
	func._ray_remote_func = ray.remote(func)
	remote_func = func._ray_remote_func
	except AttributeError:
	func_id = id(func)
	if func_id not in FunctionNode._remote_func_cache:
	FunctionNode._remote_func_cache[func_id] = ray.remote(func)
	remote_func = FunctionNode._remote_func_cache[func_id]
	return (
	remote_func
	.options(**self._bound_options)
	.remote(*self._bound_args, **self._bound_kwargs)
	)

[gemini-code-assist]

Update the test assertions to verify the attribute-based caching on the function body instead of checking the class-level cache size.

Suggested change

	assert len(FunctionNode._remote_func_cache) == 0
	ray.get(node.execute())
	assert len(FunctionNode._remote_func_cache) == 1
	assert id(double) in FunctionNode._remote_func_cache
	# Execute again - cache size should NOT increase
	ray.get(node.execute())
	assert len(FunctionNode._remote_func_cache) == 1
	assert not hasattr(double, "_ray_remote_func")
	ray.get(node.execute())
	assert hasattr(double, "_ray_remote_func")
	# Execute again - cache should still be active
	ray.get(node.execute())
	assert hasattr(double, "_ray_remote_func")

[gemini-code-assist]

Update the test assertion to verify the attribute-based caching on the function body.

Suggested change

	# 50 executions but only 1 cache entry
	assert len(FunctionNode._remote_func_cache) == 1
	# 50 executions but only 1 cache entry on the function itself
	assert hasattr(add_one, "_ray_remote_func")

[edoakes]

I don't think this execute() functionality is actually used anywhere, probably best to deprecate/remove it

[Phucvt123]

I don't think this execute() functionality is actually used anywhere, probably best to deprecate/remove it

Thanks for the feedback! That makes sense — if execute() on the non-compiled DAG path is no longer used anywhere, removing it would be a cleaner fix than caching.

Would you prefer:

1. A deprecation warning on FunctionNode._execute_impl() / DAGNode.execute() in this PR, or
2. A full removal of the execute() path in a separate PR?

[edoakes]

@Phucvt123 let's add a deprecation warning and leave it around for a couple of releases before we remove

[edoakes]

Pushed a change to adjust the warning message. Please also update the PR title and description

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit f0bf17a. Configure here.}

[Phucvt123]

@edoakes Thanks! I have updated the PR title and description to reflect the deprecation of DAGNode.execute().

I also fixed a minor syntax error in the warning message and switched it to RayDeprecationWarning so it is visible to users. Please let me know if this looks good to you!