Incorrect results in datafusion

[franeklubi]

Describe the bug
I came upon a bug while querying my custom Parquet dataset, which causes DataFusion to produce incoherent and incorrect results.

I tested my dataset in various ways, all of which produced the desired results:

• reading parquet files using python pandas, then merging and filtering the data there
• encoding into CSV, and reading the data with DataFusion
• creating an SQLite database using the provided CSV files, and using the same queries there

To Reproduce
Steps to reproduce the behavior:

1. Download all the code and data I used for testing:

issue_data.zip

Inside there are the Parquet files and CSVs with exactly the same data (also, there's an sqlite database created from the provided CSV files).

2. Use the instructions included in README.md to reproduce the issue:

The query, that fails when querying Parquet files with datafusion-cli:

-- 1. Distinct stop names
SELECT DISTINCT stop_name FROM stop INNER JOIN trip ON tid = trip_tid WHERE line = '176' ORDER BY stop_name NULLS LAST;

Change only in where from line to trip_line produces the desired results.

Expected behavior
Should produce these 27 rows:

Bartnicza
Bazyliańska
Bolesławicka
Brzezińska
Budowlana
Choszczówka
Chłodnia
Daniszewska
Fabryka Pomp
Insurekcji
Marcelin
Marywilska-Las
Ołówkowa
PKP Płudy
PKP Żerań
Parowozowa
Pelcowizna
Polnych Kwiatów
Raciborska
Rembielińska
Sadkowska
Smugowa
Starego Dębu
Zyndrama z Maszkowic
os.Marywilska
Śpiewaków
None

Query 1 from README.md (mentioned above) produces this incorrect set of 33 rows:

+----------------------+
| stop_name            |
+----------------------+
| Bartnicza            |
| Bazyliańska          |
| Bolesławicka         |
| Brzezińska           |
| Budowlana            |
| Choszczówka          |
| Chłodnia             |
| Cygańska             |
| Czołgistów           |
| Daniszewska          |
| Fabryka Pomp         |
| Insurekcji           |
| Majerankowa          |
| Marcelin             |
| Marywilska-Las       |
| Ołówkowa             |
| PKP Falenica         |
| PKP Płudy            |
| PKP Żerań            |
| Parowozowa           |
| Pelcowizna           |
| Polnych Kwiatów      |
| Raciborska           |
| Rembielińska         |
| Rokosowska           |
| Sadkowska            |
| Smugowa              |
| Starego Dębu         |
| Zbójna Góra          |
| Zyndrama z Maszkowic |
| os.Marywilska        |
| Śpiewaków            |
|                      |
+----------------------+

Additional context
Datafusion version:

$ datafusion-cli --version
DataFusion 5.1.0

My guess
Since the Parquet files have encoded NULLs, and reading the CSV files with datafusion-cli gets rid of those, my best bet is on the usage of NULLs and some weir behavior when joining.

[franeklubi]

no_nulls.zip

In the file above I removed NULLs from Parquet files (missing strings got replaced with "NULL" and ints with 0).

Querying these files produces the desired results, which further confirms my theory, that there is something wrong with NULL handling.

[jimexist]

hi @franeklubi thanks for the detailed sharing.

to simplify bug reproduction, can you help me understand the difference between parquet and csv data? specifically:

❯ CREATE EXTERNAL TABLE stop STORED AS PARQUET LOCATION './parquets/stops';
0 rows in set. Query took 0.001 seconds.
❯ select count(*) from stop;
+-----------------+
| COUNT(UInt8(1)) |
+-----------------+
| 33254           |
+-----------------+
1 row in set. Query took 0.007 seconds.

❯ CREATE EXTERNAL TABLE stop (time TEXT, trip_tid TEXT, trip_line TEXT, stop_name TEXT) STORED AS CSV LOCATION './csvs/stop.csv';
0 rows in set. Query took 0.000 seconds.
❯ select count(*) from stop;
+-----------------+
| COUNT(UInt8(1)) |
+-----------------+
| 33255           |
+-----------------+
1 row in set. Query took 0.014 seconds.

they seem to have different number of rows.

Same thing applies to trips data.

[franeklubi]

Hi @jimexist! Thanks for the reply

The issue results from the headers kept in the CSV files. That's my bad - please remove them before testing.

I will amend the issue info, so others won't get the same problem as you

[franeklubi]

I've updated the README.md of issue_data.zip, to account for the header rows in csv files.

[jimexist]

thanks for the update.

to people reading this, i'm still trying to minimize the reproduction steps, so i guess below is a simpler statement:

CREATE EXTERNAL TABLE stop_parquet STORED AS PARQUET LOCATION './parquets/stops';

CREATE EXTERNAL TABLE stop_csv (time TEXT, trip_tid TEXT, trip_line TEXT, stop_name TEXT) STORED AS CSV LOCATION './csvs/stop.csv';

❯ select distinct stop_name from stop_csv;
+------------------------------+
| stop_name                    |
+------------------------------+
| Szczęśliwice                 |
| Wawelska                     |
| Bolesławicka                 |
...
| Ceramiczna                   |
| Czołgistów                   |
+------------------------------+
134 rows in set. Query took 0.015 seconds.

❯ select distinct stop_name from stop_parquet;
+------------------------------+
| stop_name                    |
+------------------------------+
| Milenijna                    |
| Park Praski                  |
| Wolności                     |
...
| Dzika                        |
| Budowlana                    |
| PKP Płudy                    |
| Bolesławicka                 |
| Marcelin                     |
+------------------------------+
112 rows in set. Query took 0.008 seconds.

i.e. even without join, we can tell that parquet and csv reads differently.

[jimexist]

thinking out loud: doing pandas validation:

In [10]: import pandas as pd

In [11]: csv_pd = pd.read_csv('./csvs/stop.csv')

In [12]: pq_pd = pd.read_parquet('./parquets/stops')

In [13]: csv_pd.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 33254 entries, 0 to 33253
Data columns (total 4 columns):
 #   Column     Non-Null Count  Dtype
---  ------     --------------  -----
 0   time       33254 non-null  object
 1   trip_tid   32113 non-null  float64
 2   trip_line  32126 non-null  object
 3   stop_name  705 non-null    object
dtypes: float64(1), object(3)
memory usage: 1.0+ MB

In [14]: pq_pd.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 33254 entries, 0 to 33253
Data columns (total 4 columns):
 #   Column     Non-Null Count  Dtype
---  ------     --------------  -----
 0   time       33254 non-null  datetime64[ns]
 1   trip_tid   32113 non-null  float64
 2   trip_line  32126 non-null  object
 3   stop_name  705 non-null    object
dtypes: datetime64[ns](1), float64(1), object(2)
memory usage: 1.0+ MB

and at least the stop_name columns are the same.

[jimexist]

seems like these values are absent in parquet readings:

❯ select stop_name from stop_csv except select stop_name from stop_parquet order by stop_name;
+------------------------+
| stop_name              |
+------------------------+
|                        |
| CH Promenada           |
| Centrum                |
| Ciołkosza              |
| Kanał Gocławski        |
| Marszałkowska          |
| Marysin                |
| Metro Politechnika     |
| Metro Świętokrzyska    |
| Odkryta                |
| Okularowa              |
| Poligonowa             |
| Przyczółek Grochowski  |
| Rezedowa               |
| Rozbrat                |
| Saska                  |
| Zajezdnia Ostrobramska |
| Zamieniecka            |
| pl.Bankowy             |
| pl.Konstytucji         |
| pl.Na Rozdrożu         |
| stop_name              |
| Łysakowska             |
+------------------------+
23 rows in set. Query took 0.019 seconds.

and the scanning process was invalid:

❯ select distinct stop_name from stop_parquet where stop_name = 'Odkryta';
+-----------+
| stop_name |
+-----------+
| Odkryta   |
+-----------+
1 row in set. Query took 0.007 seconds.

notice the Odkryta row would be missing in the statement below.

❯ select distinct stop_name from stop_parquet;
+------------------------------+
| stop_name                    |
+------------------------------+
|                              |
| Bystra                       |
| Dobosza                      |
| Urbanistów                   |
| Pelcowizna                   |
| PUSTELNIK                    |
| Strzeleckiego                |
| Bełdan                       |
| Kijowska                     |
| Świątynia Opatrzności Bożej  |
| Białołęka-Ratusz             |
| Armatnia                     |
| Muranowska                   |
| Rokosowska                   |
| Choszczówka                  |
| os.Marywilska                |
| Metro Stokłosy               |
| Leśnej Polanki               |
| Ćwiklińskiej                 |
| Nowodwory                    |
| Stawki                       |
| PKP Falenica                 |
| Bazyliańska                  |
| Bartnicza                    |
| ARMATNIA                     |
| Metro Ratusz-Arsenał         |
| Ćmielowska                   |
| Myśliborska                  |
| Oś Królewska                 |
| RONDO ZESŁAŃCÓW SYBERYJSKICH |
| pl.Inwalidów                 |
| Opaczewska                   |
| Rudzka                       |
| Nowolipie                    |
| Polnych Kwiatów              |
| PKP Żerań                    |
| Sarmacka                     |
| rondo Zesłańców Syberyjskich |
| Leszno                       |
| Gorzykowska                  |
| Miła                         |
| pl.Narutowicza               |
| pl.Zawiszy                   |
| Majerankowa                  |
| Daniszewska                  |
| PKP Olszynka Grochowska      |
| Stare Miasto                 |
| Cm.Wolski                    |
| Metro Stadion Narodowy       |
| Małych Dębów                 |
| Marywilska-Las               |
| Powsinek                     |
| Branickiego                  |
| Vogla                        |
| Olesin                       |
| Ceramiczna                   |
| CH Marki                     |
| Dw.Wileński                  |
| Hala Kopińska                |
| Inflancka                    |
| Klaudyny                     |
| Chłodna                      |
| Zbójna Góra                  |
| Czołgistów                   |
| Sadkowska                    |
| Insurekcji                   |
| Brzezińska                   |
| Ołówkowa                     |
| Smugowa                      |
| Dąbrówka Wiślana             |
| Żerań FSO                    |
| Kino Femina                  |
| gen.Zajączka                 |
| Wola-Ratusz                  |
| Bohomolca                    |
| Chłodnia                     |
| Starego Dębu                 |
| Raciborska                   |
| Rembielińska                 |
| Wałbrzyska-Cmentarz          |
| EC Żerań                     |
| os.Potok                     |
| Metro Księcia Janusza        |
| Dw.Gdański                   |
| Mennica                      |
| Norblin                      |
| Sienna                       |
| Gwiaździsta                  |
| Sobocka                      |
| Marymont-Potok               |
| Śpiewaków                    |
| Fabryka Pomp                 |
| METRO RATUSZ-ARSENAŁ         |
| Osiedle                      |
| Szczęśliwice                 |
| Szwedzka                     |
| pl.Starynkiewicza            |
| Wawelska                     |
| Dzika                        |
| Budowlana                    |
| PKP Płudy                    |
| Bolesławicka                 |
| Marcelin                     |
| Milenijna                    |
| Park Praski                  |
| Wolności                     |
| Smocza                       |
| pl.Wilsona                   |
| Cygańska                     |
| Parowozowa                   |
| Zyndrama z Maszkowic         |
| Parafialna                   |
+------------------------------+
112 rows in set. Query took 0.008 seconds.

Given the investigation above I believe this might be related to the single distinct to group by optimization or the hash aggregation steps.

cc @Dandandan @houqp

[alamb]

I'll try and check this out in more detail tomorrow. Thanks for the investigation @jimexist

[alamb]

I spent some time looking at the parquet data and the csv data, and it looks to me like there may be something wrong with the parquet reader.

Specifically, I just ran a query that did a select *

dump_parquet.sql:

CREATE EXTERNAL TABLE stops_parquet
STORED AS PARQUET
LOCATION '/Users/alamb/Documents/Wrong-answer-datafusion-1141/issue_data/parquets/stops';

show columns from stops_parquet;

CREATE EXTERNAL TABLE trips_parquet
STORED AS PARQUET
LOCATION '/Users/alamb/Documents/Wrong-answer-datafusion-1141/issue_data/parquets/trips';

show columns from trips_parquet;

select * from stops_parquet order by time, trip_tid, trip_line, stop_name;
select * from trips_parquet order by tid, line, base_day;

and dump_csv.sql:

CREATE EXTERNAL TABLE stops_csv (time timestamp, trip_tid bigint, trip_line TEXT, stop_name TEXT)
STORED AS CSV WITH HEADER ROW
LOCATION '/Users/alamb/Documents/Wrong-answer-datafusion-1141/issue_data/csvs/stop.csv';

show columns from stops_csv;

CREATE EXTERNAL TABLE trips_csv (tid bigint, line TEXT, base_day date)
STORED AS CSV WITH HEADER ROW
LOCATION '/Users/alamb/Documents/Wrong-answer-datafusion-1141/issue_data/csvs/trip.csv';

show columns from trips_csv;

select * from stops_csv order by time, trip_tid, trip_line, stop_name;
select * from trips_csv order by tid, line, base_day;

Like this:

~/Software/arrow-datafusion/target/debug/datafusion-cli -f dump_csv.sql  > dump_csv.txt
~/Software/arrow-datafusion/target/debug/datafusion-cli -f dump_parquet.sql  > dump_parquet.txt

The results are here:
dump_csv.txt
dump_parquet.txt

And a quick visual diff shows they aren't the same

The first few lines of dump_csv.txt look like

+---------------------+----------+-----------+------------------------------+
| time                | trip_tid | trip_line | stop_name                    |
+---------------------+----------+-----------+------------------------------+
| 2021-11-15 05:00:00 | 54761677 | N64       |                              |
| 2021-11-15 05:00:00 | 54778942 | 204       |                              |
| 2021-11-15 05:00:00 | 54788307 | 186       | RONDO ZESŁAŃCÓW SYBERYJSKICH |
| 2021-11-15 05:00:00 | 54788967 | N41       |                              |
| 2021-11-15 05:00:00 | 54788988 | N41       |                              |
| 2021-11-15 05:00:00 | 54802937 | 104       |                              |

While the first few lines of dump_parquet.txt l look like:

+---------------------+----------+-----------+------------------------------+
| time                | trip_tid | trip_line | stop_name                    |
+---------------------+----------+-----------+------------------------------+
| 2021-11-15 00:00:00 | 54761677 | N64       |                              |
| 2021-11-15 00:00:00 | 54778942 | 204       |                              |
| 2021-11-15 00:00:00 | 54788307 | 186       | Armatnia                     |
| 2021-11-15 00:00:00 | 54788967 | N41       |                              |
| 2021-11-15 00:00:00 | 54788988 | N41       |                              |
| 2021-11-15 00:00:00 | 54802937 | 104       |                              |

(note that the stop name is different)

However, when I look for that mismatched line trip_tid=54788307 in the data using pandas it does match with the csv:

Here is the raw data in csv:

$ grep 54788307 issue_data/csvs/stop.csv
2021-11-15 00:00:00,54788307,186,RONDO ZESŁAŃCÓW SYBERYJSKICH

Here is what comes out when using pandas:

Python 3.8.12 (default, Oct 13 2021, 06:42:42) 
[Clang 13.0.0 (clang-1300.0.29.3)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import pandas as pd
import pandas as pd
>>> df = pd.read_parquet('issue_data/parquets/stops/2021-11.parquet')
df = pd.read_parquet('issue_data/parquets/stops/2021-11.parquet')
>>> df.to_csv('/tmp/2021-11.csv')
df.to_csv('/tmp/2021-11.csv')
>>> 

$ grep 54788307 /tmp/2021-11.csv 
16591,2021-11-15 00:00:00,54788307.0,186,RONDO ZESŁAŃCÓW SYBERYJSKICH

[alamb]

Oh weird, but then a direct query

select * from stops_parquet where trip_tid=54788307;
+---------------------+----------+-----------+------------------------------+
| time                | trip_tid | trip_line | stop_name                    |
+---------------------+----------+-----------+------------------------------+
| 2021-11-15 00:00:00 | 54788307 | 186       | RONDO ZESŁAŃCÓW SYBERYJSKICH |
+---------------------+----------+-----------+------------------------------+

Seems to get the correct answer 🤔

[alamb]

When I just do select * the answer is not correct either:

❯ select * from stops_parquet;
+---------------------+----------+-----------+------------------------------+
| time                | trip_tid | trip_line | stop_name                    |
+---------------------+----------+-----------+------------------------------+
...
| 2021-11-15 00:00:00 | 54788307 | 186       | Armatnia                     |
...
+---------------------+----------+-----------+------------------------------+
33254 rows in set. Query took 1.553 seconds.

[alamb]

I have a smaller reproducer and am trying to narrow down where the problem is but probably won't be able to work on this until next week sometime at the earliest

In case anyone else is interested, here is the repo: repro.zip

cargo run --bin datafusion-cli -- -f repro.sql | grep 54788307

It should print out

| 2021-11-15 00:00:00 | 54788307 | 186       | RONDO ZESŁAŃCÓW SYBERYJSKICH |

But actually prints out

| 2021-11-15 00:00:00 | 54788307 | 186       | Armatnia

[alamb]

I plan to focus on this issue tomorrow