peterxcli commented on issue #24768: URL: https://github.com/apache/datafusion/issues/24768#issuecomment-5626462179
My current thought is that we should prioritize completing the spillable hash join. IIUC, one of the main goals in [DuckDB’s paper](https://www.vldb.org/pvldb/vol18/p2748-kuiper.pdf) is to avoid the performance cliff when a join exceeds memory, by keeping as much work as possible in memory and spilling only the overflow. For example, suppose a build working set (including its hash table) exceeds its budget by roughly 5%. With sufficiently fine buckets and reasonably uniform data, hybrid hashing might defer only a small fraction of probe rows. Falling back for the entire join partition potentially sorts both complete inputs thus cause "performance cliff". The actual difference would depend on bucket sizes, skew, row width, and available memory, but this is why I lean toward following the hybrid hash join direction first. On implementation, I think there are a few areas worth looking into when comparing DF with DuckDB: 1. **Buffer management and data layout** DuckDB stores its materialized build data in a row layout, with separate heap blocks for variable-sized values. Its buffer manager supports larger blocks when needed, and its deferred probe data uses a column layout. DF keeps Arrow arrays in `RecordBatch` objects and uses Arrow IPC when spilling. I’m not sure how much the difference in layout and materialization cost would affect the join, especially for wide rows and strings. I think we should start with batches owned by individual buckets and use the existing spill infrastructure. We would need to ensure that spilling a bucket actually releases its memory, since slices can share backing buffers. We would also need headroom for partitioning, serialization, and rebuilding the hash table. A new buffer manager might help eventually, but I don’t think we know yet whether it is necessary. 2. **Spilling** I’m still not very familiar with the details here, but my understanding is that DuckDB can evict eligible unpinned buffers through its buffer manager. In DF, the memory pool tracks reservations, and the operator decides what to spill when it cannot grow its reservation. So marking an operator as spillable does not automatically flush its buffers. The join would need to select buckets, write them through the existing spill machinery, release their memory, and restore them for later processing. I think comparing the costs and memory requirements of these two approaches would be useful. 3. **Radix partitioning and the hash table** I think we could borrow DuckDB’s radix partitioning and replay strategy while keeping DF’s existing `JoinHashMap` initially. The buckets would determine which build and probe rows are processed together, and we could rebuild the map for the buckets that fit. That seems like a reasonable starting point before considering a larger port of DuckDB’s implementation. I also don’t expect hybrid hashing to solve every case. A bucket dominated by one key can still exceed memory, and ordering requirements and join semantics need care. For cases where SMJ fallback helps and preserves those requirements, we could introduce a config flag to allow it. My preference would be to make hybrid spilling the main path, then add fallback for the remaining cases. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
