Skip to content

[MINOR][SQL] Move iterator.hasNext into try block in executeTask #48360

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

Closed
wants to merge 2 commits into from
Closed

[MINOR][SQL] Move iterator.hasNext into try block in executeTask #48360

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e9f990b
Move iterator.hasNex into try block
viirya Oct 5, 2024
a942699
Add more log info
viirya Oct 5, 2024
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
45 changes: 27 additions & 18 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -383,32 +383,41 @@ object FileFormatWriter extends Logging {

committer.setupTask(taskAttemptContext)

val dataWriter =
if (sparkPartitionId != 0 && !iterator.hasNext) {
// In case of empty job, leave first partition to save meta for file format like parquet.
new EmptyDirectoryDataWriter(description, taskAttemptContext, committer)
} else if (description.partitionColumns.isEmpty && description.bucketSpec.isEmpty) {
new SingleDirectoryDataWriter(description, taskAttemptContext, committer)
} else {
concurrentOutputWriterSpec match {
case Some(spec) =>
new DynamicPartitionDataConcurrentWriter(
description, taskAttemptContext, committer, spec)
case _ =>
new DynamicPartitionDataSingleWriter(description, taskAttemptContext, committer)
}
}
var dataWriter: FileFormatDataWriter = null

Utils.tryWithSafeFinallyAndFailureCallbacks(block = {
dataWriter =
if (sparkPartitionId != 0 && !iterator.hasNext) {
Copy link
Member Author

@viirya viirya Oct 5, 2024 •
edited
Loading

Choose a reason for hiding this comment

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

When I looked at a customer issue, there is the following error

org.apache.spark.shuffle.FetchFailedException: Block shuffle_1_106_21 is corrupted but checksum verification passed

which is thrown at iterator.hasNext.

But I don't see the log of aborted task. It is because hasNext is not in try block. It doesn't read a good taste.

Besides, as setupTask is called, it is safer to call abort in any error case.

// In case of empty job, leave first partition to save meta for file format like parquet.
new EmptyDirectoryDataWriter(description, taskAttemptContext, committer)
} else if (description.partitionColumns.isEmpty && description.bucketSpec.isEmpty) {
new SingleDirectoryDataWriter(description, taskAttemptContext, committer)
} else {
concurrentOutputWriterSpec match {
case Some(spec) =>
new DynamicPartitionDataConcurrentWriter(
description, taskAttemptContext, committer, spec)
case _ =>
new DynamicPartitionDataSingleWriter(description, taskAttemptContext, committer)
}
}

// Execute the task to write rows out and commit the task.
dataWriter.writeWithIterator(iterator)
dataWriter.commit()
})(catchBlock = {
// If there is an error, abort the task
dataWriter.abort()
logError(log"Job ${MDC(JOB_ID, jobId)} aborted.")
if (dataWriter != null) {
dataWriter.abort()
} else {
committer.abortTask(taskAttemptContext)
}
logError(log"Job: ${MDC(JOB_ID, jobId)}, Task: ${MDC(TASK_ID, taskId)}, " +
log"Task attempt ${MDC(TASK_ATTEMPT_ID, taskAttemptId)} aborted.")
Comment on lines +415 to +416
Copy link
Member Author

@viirya viirya Oct 5, 2024

Choose a reason for hiding this comment

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

We miss task and task attempt info previously. Adding them here.

}, finallyBlock = {
dataWriter.close()
if (dataWriter != null) {
dataWriter.close()
}
})
}

Expand Down