fix: scale default memory pool size by partition count (#6562)

## Summary

- The `FairSpillPool` divides memory evenly across spillable consumers.
With up to 8 partitions, each sort consumer was limited to ~12.5MB from
a flat 100MB pool, causing merge_insert operations with large payloads
to fail with "not enough memory to continue external sort" at very small
batch sizes (e.g. 5 rows with 1MB payloads).
- Scale the default pool size to 100MB **per partition** so each
consumer gets a reasonable allocation. Explicit `LANCE_MEM_POOL_SIZE` or
`mem_pool_size` settings are respected as-is.
- This is a partial fix — very large batches can still exhaust the
per-partition budget. A more complete fix may involve revisiting the
pool type or spilling behavior for merge_insert.

## Test plan

- [x] Added unit test `test_mem_pool_size_scales_with_partitions`
verifying pool size scales correctly
- [x] Verified with a Python repro script that merge_insert with
1MB-per-row payloads no longer fails at 5 rows (now succeeds up to ~50
rows)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: westonpace

rust/lance-datafusion/src/exec.rs

@@ -313,21 +313,22 @@ impl std::fmt::Debug for LanceExecutionOptions {
     }
 }
 
-const DEFAULT_LANCE_MEM_POOL_SIZE: u64 = 100 * 1024 * 1024;
+const DEFAULT_LANCE_MEM_POOL_SIZE_PER_PARTITION: u64 = 100 * 1024 * 1024;
 const DEFAULT_LANCE_MAX_TEMP_DIRECTORY_SIZE: u64 = 100 * 1024 * 1024 * 1024; // 100GB
 
 impl LanceExecutionOptions {
     pub fn mem_pool_size(&self) -> u64 {
+        let num_partitions = self.target_partition.unwrap_or(1) as u64;
         self.mem_pool_size.unwrap_or_else(|| {
             std::env::var("LANCE_MEM_POOL_SIZE")
                 .map(|s| match s.parse::<u64>() {
                     Ok(v) => v,
                     Err(e) => {
                         warn!("Failed to parse LANCE_MEM_POOL_SIZE: {}, using default", e);
-                        DEFAULT_LANCE_MEM_POOL_SIZE
+                        DEFAULT_LANCE_MEM_POOL_SIZE_PER_PARTITION * num_partitions
                     }
                 })
-                .unwrap_or(DEFAULT_LANCE_MEM_POOL_SIZE)
+                .unwrap_or(DEFAULT_LANCE_MEM_POOL_SIZE_PER_PARTITION * num_partitions)
         })
     }
 
@@ -1051,4 +1052,35 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn test_mem_pool_size_scales_with_partitions() {
+        let default_per_partition = DEFAULT_LANCE_MEM_POOL_SIZE_PER_PARTITION;
+
+        // No partitions specified → defaults to 1 partition
+        let opts = LanceExecutionOptions::default();
+        assert_eq!(opts.mem_pool_size(), default_per_partition);
+
+        // 4 partitions → 4x the per-partition size
+        let opts = LanceExecutionOptions {
+            target_partition: Some(4),
+            ..Default::default()
+        };
+        assert_eq!(opts.mem_pool_size(), default_per_partition * 4);
+
+        // 8 partitions → 8x the per-partition size
+        let opts = LanceExecutionOptions {
+            target_partition: Some(8),
+            ..Default::default()
+        };
+        assert_eq!(opts.mem_pool_size(), default_per_partition * 8);
+
+        // Explicit mem_pool_size is not scaled
+        let opts = LanceExecutionOptions {
+            mem_pool_size: Some(50 * 1024 * 1024),
+            target_partition: Some(8),
+            ..Default::default()
+        };
+        assert_eq!(opts.mem_pool_size(), 50 * 1024 * 1024);
+    }
 }