Python bindings create duplicated qualified fields after joining

[matthewmturner]

Describe the bug
im working on getting datafusion added to db-benchmark (#147). while putting the benchmarks together i came across an error while doing the join benchmark that i wasnt expecting. specifically the error is:

Traceback (most recent call last):
  File "datafusion/join-datafusion.py", line 72, in <module>
    df = ctx.create_dataframe([ans])
Exception: DataFusion error: Plan("Schema contains duplicate qualified field name 'ce9f0daee780e4f2796b9953bd267457c.id1'")

The test code that produced that is here:

question = "small inner on int" # q1
gc.collect()
t_start = timeit.default_timer()
ans = ctx.sql("SELECT * FROM x INNER JOIN small ON x.id1 = small.id1").collect()
shape = ans_shape(ans)
print(shape)
t = timeit.default_timer() - t_start
t_start = timeit.default_timer()
df = ctx.create_dataframe([ans])
chk = df.aggregate([], [f.sum(col("v1"))]).collect()[0].column(0)[0]
chkt = timeit.default_timer() - t_start
m = memory_usage()
write_log(task=task, data=data_name, in_rows=x_data.num_rows, question=question, out_rows=shape[0], out_cols=shape[1], solution=solution, version=ver, git=git, fun=fun, run=1, time_sec=t, mem_gb=m, cache=cache, chk=make_chk([chk]), chk_time_sec=chkt, on_disk=on_disk)
del ans
gc.collect()

if i update the sql to:

SELECT x.id1, x.id2, x.id3, x.id4, small.id4, x.id5, x.id6, x.v1, small.v2 FROM x INNER JOIN small ON x.id1 = small.id1

I get:

Traceback (most recent call last):
  File "datafusion/join-datafusion.py", line 73, in <module>
    df = ctx.create_dataframe([ans])
Exception: DataFusion error: Plan("Schema contains duplicate qualified field name 'cb53bcf8886f449c3bd2651571df185d4.id4'")

to me this looks like a bug as i think i should be able to write the query without having to alias the overlapping columns (when i alias the overlapping columns it works). for example, below is the equivalent spark query.

select * from x join small using (id1)

To Reproduce
Steps to reproduce the behavior:

Expected behavior
A clear and concise description of what you expected to happen.
I should be able to run either of the following

ans = ctx.sql("SELECT x.id1, x.id2, x.id3, x.id4, small.id4, x.id5, x.id6, x.v1, small.v2 FROM x INNER JOIN small ON x.id1 = small.id1").collect()
df = ctx.create_dataframe([ans])

ans = ctx.sql("SELECT * FROM x INNER JOIN small ON x.id1 = small.id1").collect()
df = ctx.create_dataframe([ans])

Additional context
Add any other context about the problem here.

[houqp]

is this specific to python? or also happens in the rust core?

[matthewmturner]

I believe it was just in Python but I will have to double check.

[jychen7]

I think it is just python. I try following in datafusion-cli and works

select * from test join small using (c1);

with

CREATE EXTERNAL TABLE test (
    c1  VARCHAR NOT NULL,
    c2  INT NOT NULL,
    c3  SMALLINT NOT NULL,
    c4  SMALLINT NOT NULL,
    c5  INT NOT NULL,
    c6  BIGINT NOT NULL,
    c7  SMALLINT NOT NULL,
    c8  INT NOT NULL,
    c9  BIGINT NOT NULL,
    c10 VARCHAR NOT NULL,
    c11 FLOAT NOT NULL,
    c12 DOUBLE NOT NULL,
    c13 VARCHAR NOT NULL
)
STORED AS CSV
WITH HEADER ROW
LOCATION '/testing/data/csv/aggregate_test_100.csv';

CREATE EXTERNAL TABLE small (
    c1  VARCHAR NOT NULL,
    c2  INT NOT NULL,
    c3  SMALLINT NOT NULL,
    c4  SMALLINT NOT NULL,
    c5  INT NOT NULL,
    c6  BIGINT NOT NULL,
    c7  SMALLINT NOT NULL,
    c8  INT NOT NULL,
    c9  BIGINT NOT NULL,
    c10 VARCHAR NOT NULL,
    c11 FLOAT NOT NULL,
    c12 DOUBLE NOT NULL,
    c13 VARCHAR NOT NULL
)
STORED AS CSV
WITH HEADER ROW
LOCATION '/testing/data/csv/aggregate_test_100.csv';

[houqp]

perhaps we should move this issue to the python binding repo then :)

[matthewmturner]

perhaps we should move this issue to the python binding repo then :)

Yes that was my plan. Currently away and will make new issue there, with more details, in a few days.

[jychen7]

Actually, seems the problem is neither datafusion or pydatafusion, just different expection.

datafusion allow duplicate columns in ans, but pydatafusion will raise error when create_dataframe when input columns are duplicated.
e.g. select x.c2, y.c2 from x join y using (c1) limit 1 shows

+----+----+
| c2 | c2 |
+----+----+
| 1  | 1  |
+----+----+

1. I test in MySQL 5.6 and PostgreSQL 9.6, they also show duplicate column names in output, e.g. http://sqlfiddle.com/#!9/a6c585/237251 and http://sqlfiddle.com/#!17/bf2fd/25993. This also align with "All bare column field names MUST not contain relation/table qualifier." in https://arrow.apache.org/datafusion/specification/output-field-name-semantic.html
2. dataframe is correct too, since it is not expected to have duplicate input name. I believe the internal dataframe using in ctx.sql have different column name like x.c2 and y.c2, but we trucate that when output.

Given Arrow is in-memory dataframe, with extra parameter, it should be possible to ask ctx.sql to return dataframe directly? so we don't need to ctx.create_dataframe?

[matthewmturner]

@jychen7 thank you for the analysis. i am going to look into this more tomorrow.

[matthewmturner]

Im at odds here as to what expected behavior should be. The point you mentioned "All bare column field names MUST not contain relation/table qualifier." in https://arrow.apache.org/datafusion/specification/output-field-name-semantic.html seems to prevent creating dataframe from output of something like SELECT * FROM x JOIN y on x.id1 = y.id1 - which i think we should be able to do.

@houqp do you have an opinion on this?

[houqp]

@matthewmturner this is indeed expected behavior. If you need to create a dataframe from record batches, you need to explicitly create aliases to make sure all output field names are unique. Postgres behave the same way as well, see this subquery as an example: http://sqlfiddle.com/#!17/bf2fd/26130.

Given Arrow is in-memory dataframe, with extra parameter, it should be possible to ask ctx.sql to return dataframe directly?

@jychen7 , SQL queries only returns record batches, which doesn't have the concept of column qualifiers (table name). This is also to align the behavior with other dbs and query engines. For example, postgres strips all qualifiers in its query output as well. The create_dataframe API is actually not a very well named method, what it does actually is to create a memtable and wrap it with a TableScan plan. It won't make a lot sense for a sql query to return a table (memtable) in this case.

[matthewmturner]

@houqp just one follow up - postgres does allow the following:

select * from jsonData as t1
join jsonData as t2 on t1.id = t2.id

http://sqlfiddle.com/#!17/bf2fd/26200

so it would seem datafusion doesnt match that behaviour as we cant call select * after joining tables.

[houqp]

so it would seem datafusion doesnt match that behaviour as we cant call select * after joining tables.

Yeah, i think this is a subquery support bug that needs to be fixed.

[matthewmturner]

@houqp how would this work in datafusions case though since we dont allow relation / table qualifier? Postgres seems to show duplicate columns without a qualifier which i dont think we can do.

Also, regarding that being a subquery bug - would you be able to expand on that point? I hadnt really thought of this as a subquery. Are you referring to what this turns into at the logical / physical plan level.

[mesejo]

I can no longer reproduce this issue with the current version of DataFusion (28.0.0):

import datafusion
import pandas as pd

x = pd.DataFrame(data={'id1': [1, 2, 4, 5, 6], 'col2': [3, 4, 3, 5, 2], 'col3': [3, 4, 1, 2, 3]})
x.to_csv('x.csv', index=False)

x = pd.DataFrame(data={'id1': [1, 2, 4, 5, 6], 'col2': [3, 4, 3, 5, 2], 'col3': [5, 6, 7, 8, 9]})
x.to_csv('small.csv', index=False)


ctx = datafusion.SessionContext()
ctx.register_csv(name="x", path="x.csv")
ctx.register_csv(name="small", path="small.csv")


df = ctx.sql("SELECT * FROM x INNER JOIN small ON small.id1 = x.id1")
df.show()

Output

 DataFrame()
+-----+------+------+-----+------+------+
| id1 | col2 | col3 | id1 | col2 | col3 |
+-----+------+------+-----+------+------+
| 2   | 4    | 4    | 2   | 4    | 6    |
| 1   | 3    | 3    | 1   | 3    | 5    |
| 5   | 5    | 2    | 5   | 5    | 8    |
| 4   | 3    | 1    | 4   | 3    | 7    |
| 6   | 2    | 3    | 6   | 2    | 9    |
+-----+------+------+-----+------+------+

However, when I try to transform it into a pandas DataFrame, I do get an error:

df.to_pandas()

Error

Traceback (most recent call last):
  File "bug.py", line 19, in <module>
    df.to_pandas()
  File "pyarrow/array.pxi", line 837, in pyarrow.lib._PandasConvertible.to_pandas
  File "pyarrow/table.pxi", line 4114, in pyarrow.lib.Table._to_pandas
  File "/lib/python3.10/site-packages/pyarrow/pandas_compat.py", line 819, in table_to_blockmanager
    columns = _deserialize_column_index(table, all_columns, column_indexes)
  File "/lib/python3.10/site-packages/pyarrow/pandas_compat.py", line 938, in _deserialize_column_index
    columns = _flatten_single_level_multiindex(columns)
  File "/lib/python3.10/site-packages/pyarrow/pandas_compat.py", line 1184, in _flatten_single_level_multiindex
    raise ValueError('Found non-unique column index')
ValueError: Found non-unique column index

This behavior is weird; internally DataFusion knows that the columns come from different schemas. This works

df.select(col("small.id1"), col("small.col2"), col("small.col3")).collect()

Perhaps is worth trying to qualify overlapped names when transforming to a pandas DataFrame?