[Spark] Rebalance file listing across tasks in CONVERT TO DELTA schema inference - #7570
Open
cravani wants to merge 1 commit into
Open
[Spark] Rebalance file listing across tasks in CONVERT TO DELTA schema inference#7570cravani wants to merge 1 commit into
cravani wants to merge 1 commit into
Conversation
…st (e.g. a native Unity Catalog parquet table, which cannot use catalog partitions) reads every source file's Parquet footer to infer the schema. recursiveListDirs parcels those files by top-level directory, so a single large partition directory becomes one straggler task that reads all of its footers alone while the other cores idle. Rebalance the listed files across tasks by file count (footer reads are ~constant cost per file) before reading footers, gated by an internal conf spark.databricks.delta.convert.rebalanceFileListing (default true). This removes the single-directory straggler and lets cluster parallelism apply to the footer reads. It does not change the resulting table schema, AddFiles, or data. Signed-off-by: cravani <chiran54321@gmail.com>
cravani
force-pushed
the
convert-rebalance-file-listing
branch
from
August 31, 2026 22:56
ff0f21c to
8837f89
Compare
foss-contributor
added
delta-ddl
delta-spark
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 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 Delta project/connector is this regarding?
Description
CONVERT TO DELTAinfers the table schema by reading the Parquet footer of every source file. On theManualListingFileManifestpath (used when the table cannot use catalog partitions, e.g. native Unity Catalog parquet tables),DeltaFileOperations.recursiveListDirsdistributes the listed files across tasks by top-level directory. When one partition directory holds most of the files (e.g. a dominantcountry=USpartition), that directory's footer reads all run in a single skewed task while the rest of the cluster is idle, the dominant cost of converting a many-file table.This PR rebalances the listed files across tasks by file count (a footer read is ~constant cost per file) with a round-robin repartition inserted between the recursive listing and the footer reads in
ManualListingFileManifest.allFilesIt is gated by a new internal confspark.databricks.delta.convert.rebalanceFileListing(defaulttrue; set tofalseto restore the previous per-directory distribution).Effect: the single-directory skew is eliminated and footer-read work scales with cluster parallelism.
The converted table's
schema,AddFileactions, and the data are unchanged.How was this patch tested?
New unit test
ConvertToDeltaListingRebalanceSuitebuilds a skewed layout (onecountry=USdirectory with45of60files) and checks the per-partition file-count distribution ofallFileswith the confOFFvsON:45files (the skew):[45, 15].[8,8,8,8,7,7,7,7], max 8.Does this PR introduce any user-facing changes?
Not by default, added new config as a fall back when
spark.databricks.delta.convert.rebalanceFileListingis set to false.