default FileSource::supports_repartitioning to false

[jayshrivastava]

Describe the bug

This seems like an odd thing to default to true. I don't know any formats other than csv which would benefit.

I had a regression when upgrading to df52 where my files (which are a custom format) were getting split up randomly and it took a while to realize this new setting was causing it.

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[alchemist51]

Hi @jayshrivastava I don't think it's a bug, formats like parquet also uses repartitioning strategy and it helps in making the query faster. In case you don't want it for your custom format, you can override it to false?

[AdamGS]

Vortex also benefits. I think it makes sense to default to true as it made that change not breaking, not sure what code path it caused your format to hit but I think that's an unexpected outcome.

[jayshrivastava]

Sure, feel free to close out the issue if you think that makes sense.

One follow up - do you know how split points are determined? I would expect files to not be split up unless the FileSource::file_type was well known (ex. vortext, parquet, csv). My file format is none of those.

I'm also using a FileScanConfig wrapper which does not support repartitioned.

struct Wrapper<T> {
    inner: T,
    partitioning: Partitioning,
}

impl<T> DataSource for PartitionedFileScanConfig<T>
where
    T: DataSource + 'static,
{
    delegate! {
        to self.inner {
            fn open(
                &self,
                partition: usize,
                context: Arc<TaskContext>,
            ) -> Result<SendableRecordBatchStream>;

            fn eq_properties(&self) -> EquivalenceProperties;
            fn scheduling_type(&self) -> SchedulingType;
            fn partition_statistics(&self, partition: Option<usize>) -> Result<Statistics>;
            fn with_fetch(&self, limit: Option<usize>) -> Option<Arc<dyn DataSource>>;
            fn fetch(&self) -> Option<usize>;

            fn metrics(&self) -> ExecutionPlanMetricsSet;

            fn try_swapping_with_projection(
                &self,
                projection: &ProjectionExprs
            ) -> Result<Option<Arc<dyn DataSource>>>;
            fn try_pushdown_filters(
                &self,
                filters: Vec<Arc<dyn PhysicalExpr>>,
                config: &ConfigOptions,
            ) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>>;
        }
    }

    fn repartitioned(
        &self,
        _target_partitions: usize,
        _repartition_file_min_size: usize,
        _output_ordering: Option<LexOrdering>,
    ) -> Result<Option<Arc<dyn DataSource>>> {
        Ok(None)
    }

    fn output_partitioning(&self) -> Partitioning {
        self.partitioning.clone()
    }

    fn as_any(&self) -> &dyn Any {
        // Delegate to inner so downcasting to FileScanConfig works
        self.inner.as_any()
    }

    fn fmt_as(&self, t: DisplayFormatType, f: &mut Formatter) -> FmtResult {
        self.inner.fmt_as(t, f)
    }
}

[AdamGS]

Its hard to say without more context, but is it possible that you do create a FileScanConfig anywhere? This code seems correct to me for the effect you're trying to achieve.

[AdamGS]

The default repartitioning logic for files lives in FileSource::repartitioned

[jayshrivastava]

Ah okay. I was running into issues because FileSource::supports_repartitioning defaults to true and due to the default implementation FileSource::repartitioned which splits based on file size.