Severe performance regression filtering dataset imported from pyarrow in duckdb>=1.3.0

[jakkes]

What happens?

Running a filter of type

WHERE x in (long list of integers)

hangs when running on a dataset imported from Arrow. This is new in DuckDB v1.3, and is also an issue in the latest build of v1.4.

To Reproduce

Script below will

1. Write some random data in hive partitioning
2. Read the data through pyarrow
3. Import it to duckdb
4. Run a filter of form "id in (0,4,8,12,16,...,9996)".

Step four hangs in v1.3 and beyond. I am not sure if it actually has anything to do with the hive partitioning.

import time
import duckdb
import pyarrow.dataset as ds
import tempfile


IDS = 10000


class measure_execution_time:
    def __init__(self, name: str):
        self._name = name
        self._start_time = None

    def __enter__(self):
        self._start_time = time.perf_counter()

    def __exit__(self, *args):
        assert self._start_time is not None
        end_time = time.perf_counter()
        elapsed_time = end_time - self._start_time
        print(f"{self._name} took {elapsed_time:.2f} seconds")



def write_data(rootdir: str):
    duckdb.sql(
        f"""
        with ids as (
            select unnest(range({IDS})) as id
        ),
        dates as (
            select unnest(range('2020-01-01'::date, '2021-01-01'::date, '1 day'::interval)) as date
        ),
        data as (
            select
                *, random() as value, random() as sort_order,
                date_part('year', date) as year,
                date_part('month', date) as month,
                date_part('day', date) as day
            from ids cross join dates
            order by sort_order
        )

        select * from data
        """
    ).to_table("tmptable")

    duckdb.sql(f"copy tmptable to '{rootdir}' (format parquet, partition_by (year, month, day))")
    duckdb.sql("drop table tmptable")


def read_data(rootdir: str):

    dataset = ds.dataset(
        rootdir,
        partitioning="hive",
        format="parquet",
    )

    # select every fourth id
    id_filter = tuple(range(0, IDS, 4))

    (
        duckdb.from_arrow(dataset)
        # This specific line is extremely slow on DuckDB v1.3
        .filter(f"id in {id_filter}")
        .to_table("tmptable")
    )


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as rootdir:
        with measure_execution_time("Writing data"):
            write_data(rootdir)

        with measure_execution_time("Reading data"):
            read_data(rootdir)

This code, when run with

pip install pyarrow==21.0.0 duckdb==1.2.2

prints

Writing data took 2.52 seconds
Reading data took 1.40 seconds

When run with

pip install pyarrow==21.0.0 duckdb==1.3.2

or

pip install pyarrow==21.0.0 duckdb==1.4.0.dev215

prints

Writing data took 1.92 seconds

and then hangs.

Specifically, it hangs on line 67,

.filter(f"id in {id_filter}")

This seems specific to when importing data from pyarrow. If I read the data using DuckDB's parquet reader, this is not an issue.

OS:

Ubuntu

DuckDB Version:

1.3.2

DuckDB Client:

Python 3.10

Hardware:

No response

Full Name:

Jakob Stigenberg

Affiliation:

Qubos Systematic

What is the latest build you tested with? If possible, we recommend testing with the latest nightly build.

I have tested with a nightly build

Did you include all relevant data sets for reproducing the issue?

Not applicable - the reproduction does not require a data set

Did you include all code required to reproduce the issue?

• Yes, I have

Did you include all relevant configuration (e.g., CPU architecture, Python version, Linux distribution) to reproduce the issue?

• Yes, I have

[Tishj]

@szarnyasg can you please move this to https://github.com/duckdb/duckdb-python ?

[Tishj]

Problem here is likely a change made to the TableFilters, gravitating towards the use of ExpressionFilter and that change isn't properly supported by the filter pushdown of arrow performed by the python client.

Hmm, this is an OptionalFilter:

Process 35362 stopped
* thread duckdb/duckdb#6, stop reason = step over
    frame #0: 0x000000010b789db0 _duckdb.cpython-311-darwin.so`duckdb::TransformFilterRecursive(filter=0x000000012c64ce90, column_ref=vector<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, true> @ 0x00000001700b5c70, timezone_config="Europe/Amsterdam", type=0x000000013c6f9d20) at pyarrow_filter_pushdown.cpp:278:8
   275          }
   276          case TableFilterType::OPTIONAL_FILTER: {
   277                  auto &optional_filter = filter.Cast<OptionalFilter>();
-> 278                  if (!optional_filter.child_filter) {
   279                          return py::none();
   280                  }
   281                  return TransformFilterRecursive(*optional_filter.child_filter, column_ref, timezone_config, type);
Target 0: (python3) stopped.
(lldb) p optional_filter.DebugToString()
(std::string) "optional: c0 IN (0, 4, 8, 12)"

So it's not deemed critical by the filter pushdown mechanism, meaning it likely has a fallback later on that handles it regardless, it's "optional" in the sense that it improves performance but doesn't impact correctness to ignore it.
Currently all OptionalFilters in the pyarrow filter pushdown implementation are ignored

[Tishj]

Actually, I'm lying, OptionalFilter is not ignored, and IN_FILTER is explicitly handled: duckdb/duckdb#16224

[Tishj]

In general I'd urge you to not use IN in this way, the optimizer doesn't have the context to rewrite this.
This sounds like it could have been id % 4 == 0 AND id <= 9996, if this is actually representative of your use-case

[jakkes]

In my use case I just have a big list of numbers.

[jakkes]

I take it you were able to reproduce the issue? If not, please let me know. This issue is rather critical to us, as it blocks us upgrading to DuckDB v1.3, and thus from exploring DuckLake too.

[evertlammerts]

Could you e.g. do:

ids = duckdb.values(*[(id,) for id in range(0, IDS, 4)]).project("* as id").set_alias("filtered_ids")
(
    duckdb.from_arrow(dataset).set_alias("dataset")
    .join(ids, condition="dataset.id=filtered_ids.id")
    .to_table("tmptable")
)

That seems to work for me:

>>> ids = duckdb.values(*[(id,) for id in range(0, 10000, 4)]).project("* as id").set_alias("filtered_ids")
>>> (
...     duckdb.from_arrow(dataset).set_alias("dataset")
...     .join(ids, condition="dataset.id=filtered_ids.id")
...     .to_table("tmptable")
... )
100% ▕██████████████████████████████████████▏ (00:00:02.32 elapsed)

In general, huge IN(...) filters are probably best avoided.

[jakkes]

I'm sure there are workarounds, but I don't really want to go through my entire codebase to fix this. It seems this is a bug introduced in a recent release, rather than a deliberate drop of functionality?

Also, when I have benchmarked this in the past, running a big IN(...) filter proved much faster than doing an inner join.

[Tishj]

I didn't run your reproduction, I just looked at the path to figure out what might be happening

import duckdb
import pyarrow as pa

arrow_table = duckdb.sql(
	f"""
	with ids as (
		select unnest(range({100})) as id
	),
	dates as (
		select unnest(range('2020-01-01'::date, '2021-01-01'::date, '1 day'::interval)) as date
	),
	data as (
		select
			*, random() as value, random() as sort_order,
			date_part('year', date) as year,
			date_part('month', date) as month,
			date_part('day', date) as day
		from ids cross join dates
		order by sort_order
	)

	select * from data
	"""
).to_arrow_table()

duckdb.from_arrow(arrow_table).filter(f"id in (0, 4, 8, 12)").show()

I wonder if somehow the filter pushdown being implemented instead of skipped is actually hurting your performance?
That would depend on how efficient arrow can execute this.

The pushdown is implemented as a bunch of OR'd comparison expressions, it could use is_in instead (https://arrow.apache.org/docs/python/generated/pyarrow.compute.is_in.html#)

[Tishj]

You could potentially verify this by replicating this filter yourself on the arrow table, then scanning the result with duckdb.

	case TableFilterType::IN_FILTER: {
		auto &in_filter = filter.Cast<InFilter>();
		ConjunctionOrFilter or_filter;
		value_set_t unique_values;
		for (const auto &value : in_filter.values) {
			if (unique_values.find(value) == unique_values.end()) {
				unique_values.insert(value);
			}
		}
		for (const auto &value : unique_values) {
			or_filter.child_filters.push_back(make_uniq<ConstantFilter>(ExpressionType::COMPARE_EQUAL, value));
		}
		return TransformFilterRecursive(or_filter, column_ref, timezone_config, type);
	}

[jakkes]

Am I understanding correctly that since v1.2.2, IN(...) filters are now being pushed down into pyarrow? And that this pushdown possibly is really inefficient?

[Tishj]

possibly really inefficient

If you read my message, you can see that I'm asking you to help confirm/deny this, so we can fix it.
I don't think your comment is very constructive to the conversation here.

[jakkes]

Not sure why you got annoyed at my comment. I am just asking if I understood your comments correctly. It seems I did.

If you want me to debug this further inside DuckDB, then you will point me to the relevant file where I should insert the code snippet above. I have never contributed to the DuckDB codebase before.

Is it here: tools/pythonpkg/src/arrow/arrow_array_stream.cpp?