Existing file ordering is lost in some cases with projections

[alamb]

Describe the bug

When upgrading to DataFusion 52 we hit a bug in our test cases where pre-sorted data was being resorted

I spent a while (with Codex's help) tracking it down, and it seems to be due to not applying the FileScanConfig projection to existing orderings.

To Reproduce

Here is a unit test. I am working on an end to end .slt test

    #[test]
    fn equivalence_properties_projection_reorders_schema() {
        // This test ensures `project_orderings` is applied even when there are no
        // partition columns: a projection reorders the schema, and the output ordering
        // is specified in projected schema indices.
        let file_schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int64, false),
            Field::new("c", DataType::Utf8, true),
        ]));
        let object_store_url = ObjectStoreUrl::parse("test:///").unwrap();
        let table_schema = TableSchema::new(Arc::clone(&file_schema), vec![]);

        let file_source: Arc<dyn FileSource> =
            Arc::new(MockSource::new(table_schema.clone()));

        let config = FileScanConfigBuilder::new(
            object_store_url.clone(),
            Arc::clone(&file_source),
        )
            .with_projection_indices(Some(vec![2, 0]))
            .unwrap()
            // Indices are in the projected schema: [c, a] -> [0, 1].
            .with_output_ordering(vec![[sort_expr("c", 0), sort_expr("a", 1)].into()])
            .build();

        let eq_properties = config.eq_properties();
        let ordering = eq_properties
            .output_ordering()
            .expect("expected output ordering");

        let first_col = ordering[0].expr.as_any().downcast_ref::<Column>().unwrap();
        assert_eq!(first_col.name(), "c");
        assert_eq!(first_col.index(), 0);

        let second_col = ordering[0].expr.as_any().downcast_ref::<Column>().unwrap();
        assert_eq!(second_col.name(), "a");
        assert_eq!(second_col.index(), 1);
    }

Expected behavior

No response

Additional context

No response

[alamb]

I tried for a while with coded to force a SortExec purely via SQL without hitting a scenario where output_ordering is already projected before eq_properties() runs (that’s the only case the project_orderings(...) fix addresses). The SQL planner seems to
always build output_ordering in the base schema, so the bug doesn’t manifest.

Here is what I tried

# Create a table ordered by (a, b, c) using inline data. A filtered ORDER BY on b
# should not introduce an extra SortExec after projection reorders the columns.
statement ok
COPY (
  VALUES
    (1, 1, 1),
    (1, 1, 2),
    (1, 2, 1),
    (1, 2, 3),
    (2, 1, 1),
    (2, 1, 2),
    (2, 2, 1)
) TO 'test_files/scratch/order/ordered_abc/part-0.csv'
STORED AS CSV;

statement ok
CREATE EXTERNAL TABLE ordered_abc (
  a INT,
  b INT,
  c INT
)
STORED AS CSV
WITH ORDER (a ASC, b ASC, c ASC)
LOCATION 'test_files/scratch/order/ordered_abc/'
OPTIONS ('format.has_header' 'false');

query TT
EXPLAIN SELECT b, a FROM ordered_abc WHERE a = 1 ORDER BY b;
----
logical_plan
01)Sort: ordered_abc.b ASC NULLS LAST
02)--Projection: ordered_abc.b, ordered_abc.a
03)----Filter: ordered_abc.a = Int32(1)
04)------TableScan: ordered_abc projection=[a, b], partial_filters=[ordered_abc.a = Int32(1)]
physical_plan
01)SortPreservingMergeExec: [b@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[b@1 as b, a@0 as a]
03)----FilterExec: a@0 = 1
04)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1, maintains_sort_order=true
05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/order/ordered_abc/part-0.csv]]}, projection=[a, b], output_ordering=[a@0 ASC NULLS LAST, b@1 ASC NULLS LAST], file_type=csv, has_header=false

statement ok
drop table ordered_abc;

[alamb]

This appears to have been introduced by

• move projection handling into FileSource #18627 / 9f725d9

There was a similar bug here in statistics

• Projection fails on TPCDS Q13 #19075

[alamb]

After more debugging, I do not think this is an actual bug. Rather it was a change from 51 where previously the FileScanConfig::output_order included any projections, and after 52 it does not

Our code was still applying the projection incorrectly.

I will fix it in our engine and document the behavior in DataFusion more clearly