Skip to content

feat: Add task result chunking support (issue #3071) #3127

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cyforkk wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat: Add task result chunking support (issue #3071) #3127

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,6 +306,19 @@ public class CoreOptions extends OptionHolder {
rangeInt(0L, Bytes.GB),
16 * Bytes.MB
);
/**
* The chunk size for task results in bytes.
* When the compressed result exceeds this size, it will be split into
* multiple properties (task_result_0, task_result_1, ...).
* Set to 0 to disable chunking.
*/
public static final ConfigOption<Integer> TASK_RESULT_CHUNK_SIZE =
new ConfigOption<>(
"task.result_chunk_size",
"The chunk size for task results in bytes. Set 0 to disable chunking.",
rangeInt(0, BytesBuffer.BYTES_LEN_MAX),
1048576
);
public static final ConfigOption<Integer> TASK_TTL_DELETE_BATCH =
new ConfigOption<>(
"task.ttl_delete_batch",
Expand Down
49 changes: 46 additions & 3 deletions hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/task/HugeTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -575,9 +575,29 @@ protected synchronized Object[] asArray() {

if (this.result != null) {
byte[] bytes = StringEncoding.compress(this.result);
checkPropertySize(bytes.length, P.RESULT);
list.add(P.RESULT);
list.add(bytes);
int chunkSize = CoreOptions.instance().get(CoreOptions.TASK_RESULT_CHUNK_SIZE);

if (chunkSize > 0 && bytes.length > chunkSize) {

@imbajin imbajin Jul 29, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

‼️ This is not the shared persisted-result path. TaskAndResultScheduler.save() stores task metadata through asArrayWithoutResult() and writes the result through unchanged HugeTaskResult.asArray(), so HStore/distributed deployments still persist one BLOB; HugeTask.set() also enforces the total result limit before this branch runs. Please center chunking in the actual result-storage abstraction and cover both local and distributed persistence, including old-format compatibility.

// Split into chunks
int numChunks = (int) Math.ceil((double) bytes.length / chunkSize);
for (int i = 0; i < numChunks; i++) {
int start = i * chunkSize;
int end = Math.min(start + chunkSize, bytes.length);
byte[] chunk = new byte[end - start];
System.arraycopy(bytes, start, chunk, 0, end - start);
checkPropertySize(chunk.length, P.chunkKey(i));
list.add(P.chunkKey(i));
list.add(chunk);
}
// Write chunk count
list.add(P.RESULT_CHUNK_COUNT);
list.add(String.valueOf(numChunks));
} else {
// Single property (original behavior)
checkPropertySize(bytes.length, P.RESULT);
list.add(P.RESULT);
list.add(bytes);
}
}

if (this.server != null) {
Expand Down Expand Up @@ -859,9 +879,32 @@ public static final class P {
public static final String RETRIES = "~task_retries";
public static final String INPUT = "~task_input";
public static final String RESULT = "~task_result";
public static final String RESULT_CHUNK_COUNT = "~task_result_chunk_count";
public static final String DEPENDENCIES = "~task_dependencies";
public static final String SERVER = "~task_server";

/**
* Get the chunk key for a given chunk index.
* @param index the chunk index, or -1 for the non-chunked result key
* @return the property key for the chunk
*/
public static String chunkKey(int index) {
if (index < 0) {
return RESULT;
}
return RESULT + "_" + index;
}

/**
* Check if a property key is a chunked result property.
* @param key the property key to check
* @return true if the key matches the chunk pattern
*/
public static boolean isChunkedProperty(String key) {
return key.startsWith(RESULT + "_") && key.length() > RESULT.length() + 1 &&
key.substring(RESULT.length() + 1).matches("\\d+");
}

private static final String[] METADATA_PROPERTIES = new String[]{
TYPE, NAME, CALLABLE, DESCRIPTION, CONTEXT, STATUS, PROGRESS,
CREATE, UPDATE, RETRIES, DEPENDENCIES, INPUT, SERVER
Expand Down
Loading