Hi Team, I exectuted the below query against a parquet file of 1 GB size, which contains 140 million rows and 14 columns in a windows machine having 64 GB RAM and 20 cores. DAFT(v0.4) takes 50 secs to execute whereas Duckdb takes only 5 secs. Is it the expected behaviour or I miss something on the setup config or need to tweak settings daft.context.set_runner_native() start_time = time.time() query = """ SELECT t3."LOT", t3."WAFER", t3."Measure_Time", ( (count() - count(t3.over_all_pat_status))/count() )* 100 as yield FROM ( SELECT t2."LOT", t2."WAFER", t2."Measure_Time", t2."DIE_X", t2."DIE_Y", Min(t2.pat_status) as over_all_pat_status FROM ( SELECT t1.*, t1.VALUE BETWEEN t1.LL AND t1.UL as pat_status FROM ( SELECT * FROM read_parquet('test.parquet') t JOIN ( SELECT PARAMETER , (avg(VALUE) - (3 * STDDEV(VALUE))) as LL, (avg(VALUE) + (3 * STDDEV(VALUE))) as UL FROM read_parquet('test.parquet') GROUP BY PARAMETER ) t0 ON t.PARAMETER = t0.PARAMETER )t1 )t2 GROUP BY "LOT", "WAFER", "Measure_Time", "DIE_X", "DIE_Y" )t3 GROUP BY "LOT", "WAFER", "Measure_Time" ORDER BY WAFER """ df =daft.sql(query).collect() end_time = time.time() print("Execution Time : ", (end_time - start_time), "sec")

Could you share the query plan from df.explain(show_all=True)? I see a couple of nested queries, so I want to confirm if we manage to unnest them

* Source: | Number of partitions = 1 | Output schema = LOT#Utf8, WAFER#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | yield#Float64 However here is the logical plan used to produce this result: == Unoptimized Logical Plan == * Sort: Sort by = (col(WAFER), ascending, nulls last) | * Project: col(LOT), col(WAFER), col(Measure_Time), col(yield) | * Aggregation: [[count(col(LOT), All) - count(col(over_all_pat_status), Valid)] / | count(col(LOT), All)] * lit(100) as yield | Group by = col(LOT), col(WAFER), col(Measure_Time) | Output schema = LOT#Utf8, WAFER#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | yield#Float64 | * Project: col(LOT), col(WAFER), col(Measure_Time), col(DIE_X), col(DIE_Y), | col(over_all_pat_status) | * Aggregation: min(col(pat_status)) as over_all_pat_status | Group by = col(LOT), col(WAFER), col(Measure_Time), col(DIE_X), col(DIE_Y) | Output schema = LOT#Utf8, WAFER#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | DIE_X#Int32, DIE_Y#Int32, over_all_pat_status#Boolean | * Project: col(LOT), col(WAFER), col(PARAMETER), col(TEST_TYPE), | col(TEST_PROGRAM), col(MODULE), col(Measure_Time), col(Temperature), | col(DIE_X), col(DIE_Y), col(PRODUCT), col(UNIT), col(PASS_FAIL), col(VALUE), | col(t0.PARAMETER), col(LL), col(UL), col(VALUE) in [col(LL),col(UL)] as | pat_status | * Project: col(LOT), col(WAFER), col(PARAMETER), col(TEST_TYPE), | col(TEST_PROGRAM), col(MODULE), col(Measure_Time), col(Temperature), | col(DIE_X), col(DIE_Y), col(PRODUCT), col(UNIT), col(PASS_FAIL), col(VALUE), | col(t0.PARAMETER), col(LL), col(UL) | * Join: Type = Inner | Strategy = Auto | Left on = col(PARAMETER) | Right on = col(t0.PARAMETER) | Null equals Nulls = [false] | Output schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | PASS_FAIL#Utf8, VALUE#Float64, t0.PARAMETER#Utf8, LL#Float64, UL#Float64 |\ | * Project: col(PARAMETER) as t0.PARAMETER, col(LL), col(UL) | | | * Project: col(PARAMETER), col(LL), col(UL) | | | * Aggregation: mean(col(VALUE)) - [lit(3) * stddev(col(VALUE))] as LL, | | mean(col(VALUE)) + [lit(3) * stddev(col(VALUE))] as UL | | Group by = col(PARAMETER) | | Output schema = PARAMETER#Utf8, LL#Float64, UL#Float64 | | | * GlobScanOperator | | Glob paths = [test.parquet] | | Coerce int96 timestamp unit = Nanoseconds | | Use multithreading = true | | File schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | | PASS_FAIL#Utf8, VALUE#Float64 | | Partitioning keys = [] | | Output schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | | PASS_FAIL#Utf8, VALUE#Float64 | * GlobScanOperator | Glob paths = [test.parquet] | Coerce int96 timestamp unit = Nanoseconds | Use multithreading = true | File schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | PASS_FAIL#Utf8, VALUE#Float64 | Partitioning keys = [] | Output schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | PASS_FAIL#Utf8, VALUE#Float64 == Optimized Logical Plan == * Sort: Sort by = (col(WAFER), ascending, nulls last) | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } | * Project: col(LOT), col(WAFER), col(Measure_Time), [[col(LOT.local_count(All)) | - col(over_all_pat_status.local_count(Valid))] / col(LOT.local_count(All))] * | lit(100) as yield | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } | * Aggregation: count(col(LOT), All) as LOT.local_count(All), | count(col(over_all_pat_status), Valid) as over_all_pat_status.local_count(Valid) | Group by = col(LOT), col(WAFER), col(Measure_Time) | Output schema = LOT#Utf8, WAFER#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | LOT.local_count(All)#UInt64, over_all_pat_status.local_count(Valid)#UInt64 | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } | * Project: col(LOT), col(over_all_pat_status), col(WAFER), col(Measure_Time) | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } | * Aggregation: min(col(pat_status)) as over_all_pat_status | Group by = col(LOT), col(WAFER), col(Measure_Time), col(DIE_X), col(DIE_Y) | Output schema = LOT#Utf8, WAFER#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | DIE_X#Int32, DIE_Y#Int32, over_all_pat_status#Boolean | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } | * Project: col(VALUE) in [col(LL),col(UL)] as pat_status, col(LOT), col(WAFER), | col(Measure_Time), col(DIE_X), col(DIE_Y) | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } | * Join: Type = Inner | Strategy = Auto | Left on = col(PARAMETER) | Right on = col(t0.PARAMETER) | Null equals Nulls = [false] | Output schema = PARAMETER#Utf8, LOT#Utf8, WAFER#Utf8, | Measure_Time#Timestamp(Nanoseconds, None), DIE_X#Int32, DIE_Y#Int32, | VALUE#Float64, t0.PARAMETER#Utf8, LL#Float64, UL#Float64 | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = 0, | Upper bound bytes = 452014383 } |\ | * Project: col(PARAMETER) as t0.PARAMETER, col(VALUE.local_mean()) - | | col((Literal(Int64(3)) * VALUE.local_stddev())) as LL, col(VALUE.local_mean()) | | + col((Literal(Int64(3)) * VALUE.local_stddev())) as UL | | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | | 452014383, Upper bound bytes = 452014383 } | | | * Project: col(PARAMETER) as PARAMETER, col(VALUE.local_mean()) as | | VALUE.local_mean(), lit(3) * col(VALUE.local_stddev()) as (Literal(Int64(3)) * | | VALUE.local_stddev()) | | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | | 452014383, Upper bound bytes = 452014383 } | | | * Aggregation: mean(col(VALUE)) as VALUE.local_mean(), stddev(col(VALUE)) as | | VALUE.local_stddev() | | Group by = col(PARAMETER) | | Output schema = PARAMETER#Utf8, VALUE.local_mean()#Float64, | | VALUE.local_stddev()#Float64 | | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | | 452014383, Upper bound bytes = 452014383 } | | | * Project: col(VALUE), col(PARAMETER) | | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | | 452014383, Upper bound bytes = 452014383 } | | | * Num Scan Tasks = 1 | | File schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | | PASS_FAIL#Utf8, VALUE#Float64 | | Partitioning keys = [] | | Projection pushdown = [VALUE, PARAMETER] | | Output schema = PARAMETER#Utf8, VALUE#Float64 | | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | | 452014383, Upper bound bytes = 452014383 } | * Project: col(PARAMETER), col(LOT), col(WAFER), col(Measure_Time), col(DIE_X), | col(DIE_Y), col(VALUE) | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | 271608643, Upper bound bytes = 271608643 } | * Num Scan Tasks = 1 | File schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, TEST_TYPE#Utf8, | TEST_PROGRAM#Utf8, MODULE#Utf8, Measure_Time#Timestamp(Nanoseconds, None), | Temperature#Float64, DIE_X#Int32, DIE_Y#Int32, PRODUCT#Utf8, UNIT#Utf8, | PASS_FAIL#Utf8, VALUE#Float64 | Partitioning keys = [] | Projection pushdown = [PARAMETER, LOT, WAFER, Measure_Time, DIE_X, DIE_Y, VALUE] | Filter pushdown = [not(is_null(col(PARAMETER))) & not(is_null(col(PARAMETER)))] | & not(is_null(col(PARAMETER))) | Output schema = LOT#Utf8, WAFER#Utf8, PARAMETER#Utf8, | Measure_Time#Timestamp(Nanoseconds, None), DIE_X#Int32, DIE_Y#Int32, | VALUE#Float64 | Stats = { Lower bound rows = 0, Upper bound rows = None, Lower bound bytes = | 271608643, Upper bound bytes = 271608643 }

Also, we have a really nifty tool for analyzing query performance. Could you run your query with the explain analyze environment variable?

DAFT_DEV_ENABLE_EXPLAIN_ANALYZE=1

Nice! Here’s a visualization after pasting it into

<https://mermaid.live>

Nice Visualization !!!

Interestingly the times seems to be really really fast for computation. I’m guessing most of the time here is being spent on the Scan of Parquet. Is this data sensitive? Would love to poke around the data and query if you’re open to sharing it!

Sure Jay, I will share it shortly

The cpu times for explain analyze needs fixing, PR here. https://github.com/Eventual-Inc/Daft/pull/3511 . Regardless, it showed that the hash join build sink processed 143 M rows, and the probe was only 496 rows. This is probably the bottleneck

Also, I could observe that DuckDB takes up to 3 GB Memory whereas Daft takes 25 GB to execute the query. This is the peak memory consumption during query execution.

Can confirm that switching the join side works. Runtime reduced from 35.2s -> 4.4s. Currently the join side is determined by comparing the size bytes for each side, using the row count instead leads to choosing the other side for the join. Will have a PR out for this soon

Here are results using Daft 0.4.3, Could see improvement in timings, but not as you get

@Colin Ho could you chime in here on join-side selection?

I actually ran the same query on duckdb and daft on my laptop (12 core, 36gb macbook pro), and i get daft: 4.40s duckdb: 1.89s The ratio looks almost the same as yours, so I think it is probably just the difference in the hardware.

@Colin Ho I tried it on my laptop (20 core, 64GB, Windows 10, 12th Gen Intel(R) Core(TM) i9-12900H 2.50 GHz Processor), Daft 0.4.1.3 is four times slower than DuckDB. I had run it with default Daft settings

Yes, that's likely the best Daft can do as of today. Looking at the profile, the biggest time consumers are the grouped aggregate and hash join probe. We would need to optimize those kernels and make them faster