-
Notifications
You must be signed in to change notification settings - Fork 2.1k
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.
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
Spark 3.3: Change default distribution modes #6828
Spark 3.3: Change default distribution modes #6828
Conversation
} else if (table.spec().isPartitioned()) { | ||
return HASH; | ||
} else { | ||
return NONE; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
if we keep this as hash, we will avoid small files even with unpartitioned tables by forcing the rebalance right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We have to request a valid Distribution
in order for AQE to do its job. If the table is unpartitioned, I don't think we have what to cluster by.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is purely for inserts, though. I should probably call it defaultInsertDistributionMode()
or something.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I would think we could just cluster by all columns
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am not sure how that would perform, to be honest. What about exploring this separately? Maybe, we can also handle this natively in Spark.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What about exploring this separately?
+1, the benefit seems to be not so straightforward to cluster by all columns. We can first have this generic default and then make more changes for unpartitioned case.
@@ -491,6 +491,7 @@ public void testReadPartitionColumn() throws Exception { | |||
.option(SparkReadOptions.VECTORIZATION_ENABLED, String.valueOf(vectorized)) | |||
.load(baseLocation) | |||
.select("struct.innerName") | |||
.orderBy("struct.innerName") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Required by the check below as the default distribution changes the order of elements.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
looks good to me!
e3f9394
to
fa7965c
Compare
Thanks for reviewing, @RussellSpitzer @jackye1995 @dramaticlly! |
This PR changes the default distribution modes in Spark 3.3.
Fixes #6679.