Allow FileSource-specific repartitioning
#14754
Conversation
|
The CI failure doesn't seem related to the change |
|
While refactoring old sources into the |
There was a problem hiding this comment.
Thank you @AdamGS -- this looks great to me
Do you think we can remove supports_repartition as well in this PR?
I agree with @mertak-synnada that this is a nice API improvement
| @@ -67,4 +69,33 @@ pub trait FileSource: Send + Sync { | |||
| /// If this returns true, the DataSourceExec may repartition the data | |||
| /// by breaking up the input files into multiple smaller groups. | |||
| fn supports_repartition(&self, config: &FileScanConfig) -> bool; | |||
There was a problem hiding this comment.
It seems like we could remove supports_repartition as well:
- This API hasn't yet been released so it wouldn't be a breaking change
- It would ensure that all places in the code that need to check repartition use a single API (and thus are consistently done)
There was a problem hiding this comment.
yeah absolutely, give me a second to pull it up and take a look
| output_ordering: Option<LexOrdering>, | ||
| config: &FileScanConfig, | ||
| ) -> datafusion_common::Result<Option<FileScanConfig>> { | ||
| if config.file_compression_type.is_compressed() || config.new_lines_in_values { |
There was a problem hiding this comment.
the first one just makes sense to me as it makes it impossible (or at least very hard) to map byte ranges to actual offset in the uncompressed file. The second one is CSV only but as long as its there I think its fine to respect it here.
There was a problem hiding this comment.
I agree in general the new_lines_in_values is pretty CSV specific, but I also think this code is backwards compatible and can be overridden by sub classes now. We can revise the default value in the future if needed
| output_ordering: Option<LexOrdering>, | ||
| config: &FileScanConfig, | ||
| ) -> datafusion_common::Result<Option<FileScanConfig>> { | ||
| if config.file_compression_type.is_compressed() || config.new_lines_in_values { |
There was a problem hiding this comment.
I agree in general the new_lines_in_values is pretty CSV specific, but I also think this code is backwards compatible and can be overridden by sub classes now. We can revise the default value in the future if needed
| || config.new_lines_in_values | ||
| || self.as_any().downcast_ref::<AvroSource>().is_some()) | ||
|
|
||
| fn repartitioned( |
|
THanks again @AdamGS and @xudong963 |
Which issue does this PR close?
I also want to suggest some changes around
FileGroupPartitioner/PartitionedFilebut I need to do some more thinking there and this change is useful without them IMO.Rationale for this change
Allows specific file formats to define their own file repartitioning.
What changes are included in this PR?
Move the logic in
FileScanConfig::repartitioneddown to theFileSourcetrait as the default interface, allowing externalFileSourceimplementation to control their repartitioning logic.Are these changes tested?
Are there any user-facing changes?
Introduces a new function with a default implementation to a trait.