EnforceDistribution fails, seems to turn all the types of the schema to UInt64

[fabianmurariu]

Describe the bug

This happens in 37 it works in 36
EnforceDistribution fails with "PhysicalOptimizer rule 'EnforceDistribution' failed, due to generate a different schema, original schema:

To Reproduce

WITH e1 AS (SELECT * FROM _default), e2 AS (SELECT * FROM _default), a AS (SELECT * FROM nodes), b AS (SELECT * FROM nodes), c AS (SELECT * FROM nodes) SELECT a.name, b.name, c.name FROM e1 JOIN a ON e1.src = a.id JOIN b ON e1.dst = b.id JOIN e2 ON b.id = e2.src JOIN c ON e2.dst = c.id WHERE e1.id <> e2.id

Error

called `Result::unwrap()` on an `Err` value: Context("EnforceDistribution", Internal("PhysicalOptimizer rule 'EnforceDistribution' failed, due to generate a different schema, original schema: Schema { fields: [Field { name: \"name\", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }, new schema: Schema { fields: [Field { name: \"name\", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"name\", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }"))
```


### Expected behavior

Expected optimizer rules to be skipped or not to run

### Additional context

```Projection: a.name, b.name, c.name
  Inner Join: e2.dst = c.id
    Projection: a.name, b.name, e2.dst
      Inner Join: b.id = e2.src Filter: e2.id != e1.id
        Projection: e1.id, a.name, b.id, b.name
          Inner Join: e1.dst = b.id
            Projection: e1.id, e1.dst, a.name
              Inner Join: e1.src = a.id
                SubqueryAlias: e1
                  TableScan: _default projection=[id, src, dst]
                SubqueryAlias: a
                  TableScan: nodes projection=[id, name]
            SubqueryAlias: b
              TableScan: nodes projection=[id, name]
        SubqueryAlias: e2
          TableScan: _default projection=[id, src, dst]
    SubqueryAlias: c
      TableScan: nodes projection=[id, name]
```

[alamb]

Thanks for the report @fabianmurariu

Is there any way we can get a self contained reproducer? I ran the query in the description and it doesn't seem to have all the tables

> WITH e1 AS (SELECT * FROM _default), e2 AS (SELECT * FROM _default), a AS (SELECT * FROM nodes), b AS (SELECT * FROM nodes), c AS (SELECT * FROM nodes) SELECT a.name, b.name, c.name FROM e1 JOIN a ON e1.src = a.id JOIN b ON e1.dst = b.id JOIN e2 ON b.id = e2.src JOIN c ON e2.dst = c.id WHERE e1.id <> e2.id;
Error during planning: table 'datafusion.public._default' not found

[fabianmurariu]

I'll try next week to open source the code where this is happening

[alamb]

Thanks @fabianmurariu

cc @mustafasrepo in case you have any thoughts

[mustafasrepo]

The physical plans before and after enforce distribution rule, might help in locating the problem. You can use get_plan_string helper to print this information. Putting prints at the start and at the end of

datafusion/datafusion/core/src/physical_optimizer/enforce_distribution.rs

Line 189 in 2a15614

fn optimize(

will produce necessary logs as far as I can tell.

[mustafasrepo]

Thanks @fabianmurariu

cc @mustafasrepo in case you have any thoughts

I have tried to reproduce problem by defining absolutely necessary fields in the query with below queries

statement ok
CREATE TABLE IF NOT EXISTS _default (name VARCHAR, src BIGINT, dst BIGINT, id BIGINT) AS VALUES('mustafa', 1, 2, 0),('test', 2, 3, 1);

statement ok
CREATE TABLE IF NOT EXISTS nodes (name VARCHAR,id BIGINT) AS VALUES('TR', 1),('GR', 2);

statement ok
set datafusion.execution.target_partitions = 8;

query TTT
WITH e1 AS (SELECT * FROM _default),
  e2 AS (SELECT * FROM _default),
  a AS (SELECT * FROM nodes),
  b AS (SELECT * FROM nodes),
  c AS (SELECT * FROM nodes)
  SELECT a.name, b.name, c.name
FROM e1 JOIN a ON e1.src = a.id JOIN b ON e1.dst = b.id JOIN e2 ON b.id = e2.src JOIN c ON e2.dst = c.id WHERE e1.id <> e2.id;
----

However, this test seems to pass. Unfortunately I cannot debug further. After seeing plans, or after full reproducer I will take another look.

[fabianmurariu]

Strange, I'm encountering this with custom TableProviders, I'll be able to share more next week tho