Skip to content

[SPARK-41246][core] Solve the problem of RddId negative #38781

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
yorksity wants to merge 4 commits into from
Closed

[SPARK-41246][core] Solve the problem of RddId negative #38781

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4e5a347
Solve the problem of RddId negative
yorksity Nov 24, 2022
0a7d560
Merge branch 'master' of github.com:apache/spark
yorksity Nov 24, 2022
9f97604
Merge branch 'master' of github.com:apache/spark
yorksity Dec 11, 2022
9e96549
Solve the problem of RddId negative
yorksity Dec 11, 2022
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
14 changes: 12 additions & 2 deletions core/src/main/scala/org/apache/spark/SparkContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2570,10 +2570,20 @@ class SparkContext(config: SparkConf) extends Logging {

private[spark] def newShuffleId(): Int = nextShuffleId.getAndIncrement()

private val nextRddId = new AtomicInteger(0)
private var nextRddId = new AtomicInteger(0)
Copy link

@schlosna schlosna Nov 28, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

What happens when the RDD ID overflows within a SparkContext? Are there tests that cover these cases?

Curious if it would be better to switch to an AtomicLong and just modulo max int?

Suggested change
private var nextRddId = new AtomicInteger(0)
private[spark] def newRddId(): Int = (nextRddId.getAndIncrement() % Integer.MAX_VALUE).toInt

Copy link
Contributor Author

@yorksity yorksity Dec 11, 2022

Choose a reason for hiding this comment

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

What happens when the RDD ID overflows within a SparkContext?
When BlockManager generates BlockId, BlockId only supports positive rddid, so BlockId generation fails.

switch to an AtomicLong ?
The scope of influence is very broad, which may require extensive discussion.

Copy link
Contributor

@mridulm mridulm Dec 12, 2022

Choose a reason for hiding this comment

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

You dont need to make this a var - see below.


/** Register a new RDD, returning its RDD ID */
private[spark] def newRddId(): Int = nextRddId.getAndIncrement()
private[spark] def newRddId(): Int = {
var id = nextRddId.getAndIncrement()
if (id >= 0) {
return id
}
this.synchronized {
nextRddId = new AtomicInteger(0)
id = nextRddId.getAndIncrement()
Copy link
Contributor

@vinodkc vinodkc Nov 26, 2022

Choose a reason for hiding this comment

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

Can we avoid the duplicate call of 'nextRddId.getAndIncrement()'?

}
id
}
Comment on lines +2576 to +2586
Copy link
Contributor

@mridulm mridulm Dec 12, 2022

Choose a reason for hiding this comment

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

Something like this would remove the need for synchronization, etc

Suggested change
private[spark] def newRddId(): Int = {
var id = nextRddId.getAndIncrement()
if (id >= 0) {
return id
}
this.synchronized {
nextRddId = new AtomicInteger(0)
id = nextRddId.getAndIncrement()
}
id
}
private[spark] def newRddId(): Int = {
nextRddId.getAndUpdate { i =>
var nextValue = i + 1
if (nextValue < 0) {
nextValue = 0
}
nextValue
}
}


/**
* Registers listeners specified in spark.extraListeners, then starts the listener bus.
Expand Down