Row groups are read out of order or with completely different values about arrow-datafusion HOT 4 CLOSED

Thanks for sharing and helping resolve this issue so quickly 😁

from arrow-datafusion.

Thank you for the report and the reproducer ❤️

read row groups in order they were written

This is not my expectation.

DataFusion reads row groups in parallel, potentially out of order, with multiple threads as an optimization. To preserve the order of the data you can either set the configuration datafusion.optimizer.repartition_file_scans to false or else communicate the order of the data in the files using the CREATE EXTERNAL TABLE .. WITH ORDER clause and then explicitly ask for that order in your query.

read the same values for the same row group even when the file increases in size
read the same values as the python pyarrow parquet reader

Yes I agree these are also my expectation

Maybe you can try setting datafusion.optimizer.repartition_file_scans to false and see if that makes the data consistent

from arrow-datafusion.

Setting datafusion.optimizer.repartition_file_scans to false like this fixes things. ✔️

    let session_cfg =
        SessionConfig::new().set_str("datafusion.optimizer.repartition_file_scans", "false");
    let session_ctx = SessionContext::new_with_config(session_cfg);

However, it's unclear how it interacts with other options and affects memory and performance. So here's what I have -

It is a given that the data will be sorted based on timestamp like this

    let parquet_options = ParquetReadOptions::<'_> {
        skip_metadata: Some(false),
        file_sort_order: vec![vec![Expr::Sort(Sort {
            expr: Box::new(col("ts_init")),
            asc: true,
            nulls_first: true,
        })]],
        ..Default::default()
    };

Then there are two approaches to get row groups/data in order -

• Using an order by clause in the query session_ctx.sql("SELECT * FROM data ORDER BY ts_init"). From our previous discussion, doing an order by on an already sorted column does not incur an additional overhead.

• Setting datafusion.optimizer.repartition_file_scans to false ensures that the data is read in sequential order of row groups.

It's not clear to me how each option affects the performance and memory usage. Do you have any guidance around it?

from arrow-datafusion.

Setting datafusion.optimizer.repartition_file_scans to false like this fixes things. ✔️

That is great news

Using an order by clause in the query session_ctx.sql("SELECT * FROM data ORDER BY ts_init"). From our #7675 (comment), doing an order by on an already sorted column does not incur an additional overhead.

This is the approach I would recommend (we do it in InfluxDB 3.0).

You can verify that there are no additional sorts, etc in the plan by using EXPLAIN to look at the query plan

If you use this approach, you should also be able to avoid setting datafusion.optimizer.repartition_file_scans to false as the optimizer will take care of it automatically

from arrow-datafusion.