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 2 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
17 changes: 15 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,23 @@ 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 {
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()'?

if (id < 0) {
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.

Instead of this if (id < 0) condition, else part of the previous if (id >= 0) can be used

nextRddId = new AtomicInteger(0)
id = nextRddId.getAndIncrement()
Copy link
Contributor

@vinodkc vinodkc Nov 27, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Can you try this ?

nextRddId = new AtomicInteger(1)
id = nextRddId

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.

Thank you for your review. I updated the code.

}
}
id
}

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