Apply fixes from PR #8: memory leak and dead else branch

Two bugs identified by copilot-swe-agent in PR #8:

- lib.rs: fix memory leak on ArrowArrayStreamReader::from_raw() failure.
  The original code called libc::free() after the ? early-return, so the
  malloc'd ArrowArrayStream struct leaked if from_raw() failed. Fix:
  free unconditionally before propagating the error.

- lance_writer.cpp: remove dead no-op else branch in StreamState::push().
  The else path (queue not full) called not_full_cv_.wait() with a
  predicate already satisfied, making it a guaranteed immediate return.
  Removing it clarifies the backpressure logic.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tsafin

src/writers/lance_writer.cpp

@@ -52,10 +52,6 @@ struct StreamState {
             auto waited = std::chrono::duration_cast<std::chrono::nanoseconds>(wait_end - wait_start).count();
             stall_ns_.fetch_add(static_cast<uint64_t>(waited), std::memory_order_relaxed);
             stall_count_.fetch_add(1, std::memory_order_relaxed);
-        } else {
-            not_full_cv_.wait(lock, [&] {
-                return closed_ || queue_.size() < max_queue_batches_ || !status_.ok();
-            });
         }
         if (!status_.ok()) {
             return;

third_party/lance-ffi/src/lib.rs

@@ -211,9 +211,9 @@ impl LanceWriterHandle {
                     return Err("Null ArrowArrayStream".to_string());
                 }
 
-                let reader = unsafe { ArrowArrayStreamReader::from_raw(stream_ptr) }
-                    .map_err(|e| format!("Failed to import ArrowArrayStream: {}", e))?;
+                let result = unsafe { ArrowArrayStreamReader::from_raw(stream_ptr) };
                 unsafe { libc::free(stream_ptr as *mut c_void) };
+                let reader = result.map_err(|e| format!("Failed to import ArrowArrayStream: {}", e))?;
 
                 let compressed_schema = Arc::new(apply_compression_metadata(reader.schema().as_ref()));
                 let compression_reader = CompressionReader::new(reader, compressed_schema);