minor: include taskAttemptId in log messages #2467
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
andygrove
changed the title
acquireMemory returns less memory than requestedacquireMemory returns less memory than requested [WIP]
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #2467 +/- ##
============================================
+ Coverage 56.12% 58.33% +2.20%
- Complexity 976 1438 +462
============================================
Files 119 146 +27
Lines 11743 13518 +1775
Branches 2251 2350 +99
============================================
+ Hits 6591 7886 +1295
- Misses 4012 4400 +388
- Partials 1140 1232 +92 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
andygrove
changed the title
acquireMemory returns less memory than requested [WIP]
andygrove
changed the title
comphead
approved these changes
Oct 1, 2025
| if used != 0 { | ||
| warn!("CometUnifiedMemoryPool dropped with {used} bytes still reserved"); | ||
| warn!( | ||
| "Task {} dropped CometUnifiedMemoryPool with {used} bytes still reserved", |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} dropped CometUnifiedMemoryPool with {used} bytes still reserved", | |
| "Task ID {} dropped CometUnifiedMemoryPool with {used} bytes still reserved", |
| .unwrap_or_else(|_| panic!("Failed to release {size} bytes")); | ||
| if let Err(e) = self.release_to_spark(size) { | ||
| panic!( | ||
| "Task {} failed to return {size} bytes to Spark: {e:?}", |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} failed to return {size} bytes to Spark: {e:?}", | |
| "Task ID {} failed to return {size} bytes to Spark: {e:?}", |
Member
Author
There was a problem hiding this comment.
I was following Spark's convention for logging so that we can use grep to search for Task 1234 and view the combined logging from both Spark and Comet.
~/git/apache/apache-spark-3.5.6$ find . -name "*.scala" -exec grep taskAttemptId {} \; | grep log
logDebug(s"Starting pushing blocks for the task ${context.taskAttemptId()}")
logWarning(s"Task ${taskAttemptId.get} already completed, not releasing lock for $blockId")
logTrace(s"Task $taskAttemptId trying to acquire read lock for $blockId")
logTrace(s"Task $taskAttemptId acquired read lock for $blockId")
logTrace(s"Task $taskAttemptId trying to acquire write lock for $blockId")
logTrace(s"Task $taskAttemptId acquired write lock for $blockId")
logTrace(s"Task $taskAttemptId downgrading write lock for $blockId")
logTrace(s"Task $taskAttemptId releasing lock for $blockId")
logTrace(s"Task $taskAttemptId trying to remove block $blockId")
logInfo(s"Task ${TaskContext.get().taskAttemptId} force spilling in-memory map to disk " +
logInfo(s"Task ${context.taskAttemptId} force spilling in-memory map to disk and " +
logInfo(s"TID $taskAttemptId waiting for at least 1/2N of $poolName pool to be free")
logInfo(s"Task $taskAttemptId from Stage $stageId(Attempt $stageAttemptNumber) has entered " +
logInfo(s"Task $taskAttemptId from Stage $stageId(Attempt $stageAttemptNumber) waiting " +
logInfo(s"Task $taskAttemptId from Stage $stageId(Attempt $stageAttemptNumber) finished " +
logInfo(s"Task $taskAttemptId from Stage $stageId(Attempt $stageAttemptNumber) failed " +
``
| { | ||
| panic!("overflow when releasing {size} of {prev} bytes"); | ||
| panic!( | ||
| "Task {} overflow when releasing {size} of {prev} bytes", |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} overflow when releasing {size} of {prev} bytes", | |
| "Task ID {} overflow when releasing {size} of {prev} bytes", |
|
|
||
| return Err(resources_datafusion_err!( | ||
| "Failed to acquire {} bytes, only got {}. Reserved: {}", | ||
| "Task {} failed to acquire {} bytes, only got {}. Reserved: {}", |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} failed to acquire {} bytes, only got {}. Reserved: {}", | |
| "Task ID {} failed to acquire {} bytes, only got {}. Reserved: {}", |
| { | ||
| return Err(resources_datafusion_err!( | ||
| "Failed to acquire {} bytes due to overflow. Reserved: {}", | ||
| "Task {} failed to acquire {} bytes due to overflow. Reserved: {}", |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} failed to acquire {} bytes due to overflow. Reserved: {}", | |
| "Task ID {} failed to acquire {} bytes due to overflow. Reserved: {}", |
| // Returns the actual amount of memory (in bytes) granted. | ||
| public long acquireMemory(long size) { | ||
| if (logger.isTraceEnabled()) { | ||
| logger.trace("Task {} requested {} bytes", taskAttemptId, size); |
Contributor
There was a problem hiding this comment.
Suggested change
| logger.trace("Task {} requested {} bytes", taskAttemptId, size); | |
| logger.trace("Task ID {} requested {} bytes", taskAttemptId, size); |
| long newUsed = used.addAndGet(acquired); | ||
| if (acquired < size) { | ||
| logger.warn( | ||
| "Task {} requested {} bytes but only received {} bytes. Current allocation is {} and " |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} requested {} bytes but only received {} bytes. Current allocation is {} and " | |
| "Task ID {} requested {} bytes but only received {} bytes. Current allocation is {} and " |
| // Called by Comet native through JNI | ||
| public void releaseMemory(long size) { | ||
| if (logger.isTraceEnabled()) { | ||
| logger.trace("Task {} released {} bytes", taskAttemptId, size); |
Contributor
There was a problem hiding this comment.
Suggested change
| logger.trace("Task {} released {} bytes", taskAttemptId, size); | |
| logger.trace("Task ID {} released {} bytes", taskAttemptId, size); |
| if (newUsed < 0) { | ||
| logger.error( | ||
| "Used memory is negative: " + newUsed + " after releasing memory chunk of: " + size); | ||
| "Task {} used memory is negative ({}) after releasing {} bytes", |
Contributor
There was a problem hiding this comment.
Suggested change
| "Task {} used memory is negative ({}) after releasing {} bytes", | |
| "Task ID {} used memory is negative ({}) after releasing {} bytes", |
coderfender
pushed a commit
to coderfender/datafusion-comet
that referenced
this pull request
Dec 13, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #2453
Rationale for this change
This additional information in log messages and exceptions has been helpful to me in tracking down memory issues. I think it could be helpful in the future.
What changes are included in this PR?
How are these changes tested?