nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: python-zk/kazoo
Pull request #497 (2018-03-15 22:40:28)

fix(core): Correctly fire multiple callbacks
Due to the use of unbound lambdas in AsyncResult, multiple callbacks
wouldn't actually get called. it would only call the final callback N
times, where N is the number of registered callbacks.

packysauce: aaaaand a test timed out. The rest are green though!
packysauce: Thanks for kicking the tests!
jeffwidman: There are merge conflicts, so can you rebase? I just tried doing it, but I don't have access rights to your branch... (there's a small checkbox on the side when you submit a PR to allow maintainers to commit to your PR branch)
jeffwidman: @packysauce can you rebase? I'd like to merge this
packysauce: Sorry about that! 
jeffwidman: Can you rebase rather than merging in master? It keeps the commit tree clean. Thanks!
packysauce: holy moly that was an adventure! I think I got it now though

Patch:
diff --git a/kazoo/handlers/utils.py b/kazoo/handlers/utils.py
index e889f82..f18f3ac 100644
--- a/kazoo/handlers/utils.py
+++ b/kazoo/handlers/utils.py
@@ -46,7 +46,7 @@ class AsyncResult(object):
             self._exception = None
             for callback in self._callbacks:
                 self._handler.completion_queue.put(
-                    lambda: callback(self)
+                    functools.partial(callback, self)
                 )
             self._condition.notify_all()
 
@@ -56,7 +56,7 @@ class AsyncResult(object):
             self._exception = exception
             for callback in self._callbacks:
                 self._handler.completion_queue.put(
-                    lambda: callback(self)
+                    functools.partial(callback, self)
                 )
             self._condition.notify_all()
 
@@ -103,7 +103,7 @@ class AsyncResult(object):
             # Are we already set? Dispatch it now
             if self.ready():
                 self._handler.completion_queue.put(
-                    lambda: callback(self)
+                    functools.partial(callback, self)
                 )
                 return
 


Test patch:
diff --git a/kazoo/tests/test_threading_handler.py b/kazoo/tests/test_threading_handler.py
index 2444533..a6b6f3a 100644
--- a/kazoo/tests/test_threading_handler.py
+++ b/kazoo/tests/test_threading_handler.py
@@ -376,3 +376,19 @@ class TestThreadingAsync(unittest.TestCase):
         assert regular_function() == 'hello'
         assert mock_handler.completion_queue.put.called
         assert async_result.get() == 'hello'
+
+    def test_multiple_callbacks(self):
+        mockback1 = mock.Mock(name='mockback1')
+        mockback2 = mock.Mock(name='mockback2')
+        handler = self._makeHandler()
+        handler.start()
+
+        async_result = self._makeOne(handler)
+        async_result.rawlink(mockback1)
+        async_result.rawlink(mockback2)
+        async_result.set('howdy')
+        async_result.wait()
+        handler.stop()
+
+        mockback2.assert_called_once_with(async_result)
+        mockback1.assert_called_once_with(async_result)
Repository: python/bedevere
Pull request #563 (2023-05-22 21:32:17)

Update label creation
Add the label description upon creation.


Patch:
diff --git a/bedevere/backport.py b/bedevere/backport.py
index 42b7041..dbe074d 100644
--- a/bedevere/backport.py
+++ b/bedevere/backport.py
@@ -130,5 +130,6 @@ async def maintenance_branch_created(event, gh, *args, **kwargs):
     if MAINTENANCE_BRANCH_RE.match(branch_name):
         await gh.post(
             "/repos/python/cpython/labels",
-            data={"name": f"needs backport to {branch_name}", "color": "c2e0c6"},
+            data={"name": f"needs backport to {branch_name}", "color": "c2e0c6",
+                  "description": "bug and security fixes"},
         )


Test patch:
diff --git a/tests/test_backport.py b/tests/test_backport.py
index d148ec4..0da9771 100644
--- a/tests/test_backport.py
+++ b/tests/test_backport.py
@@ -452,7 +452,8 @@ async def test_maintenance_branch_created(ref):
     await backport.router.dispatch(event, gh)
     label_creation_post = gh.post_[0]
     assert label_creation_post[0] == "https://api.github.com/repos/python/cpython/labels"
-    assert label_creation_post[1] == {'name': f"needs backport to {ref}", 'color': 'c2e0c6'}
+    assert label_creation_post[1] == {'name': f"needs backport to {ref}", 'color': 'c2e0c6',
+                                      'description': 'bug and security fixes'}
 
 
 @pytest.mark.parametrize('ref', ['backport-3.9', 'test', 'Mariatta-patch-1'])
Repository: polarsignals/frostdb
Pull request #582 (2023-11-03 13:39:16)

db: add GetDB and DropDB
This allows checking whether a DB exists, and dropping a DB.


Patch:
diff --git a/db.go b/db.go
index c91550d..e2b2aad 100644
--- a/db.go
+++ b/db.go
@@ -599,6 +599,30 @@ func (s *ColumnStore) DB(ctx context.Context, name string, opts ...DBOption) (*D
 	return db, nil
 }
 
+func (s *ColumnStore) GetDB(name string) (*DB, error) {
+	s.mtx.RLock()
+	defer s.mtx.RUnlock()
+	db, ok := s.dbs[name]
+	if !ok {
+		return nil, fmt.Errorf("db %s not found", name)
+	}
+	return db, nil
+}
+
+func (s *ColumnStore) DropDB(name string) error {
+	db, err := s.GetDB(name)
+	if err != nil {
+		return err
+	}
+	if err := db.Close(WithClearStorage()); err != nil {
+		return err
+	}
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	delete(s.dbs, name)
+	return nil
+}
+
 func (db *DB) openWAL(ctx context.Context) (WAL, error) {
 	wal, err := wal.Open(
 		db.logger,


Test patch:
diff --git a/db_test.go b/db_test.go
index 6aa69b3..a4ba78e 100644
--- a/db_test.go
+++ b/db_test.go
@@ -1875,7 +1875,8 @@ func Test_DB_DropStorage(t *testing.T) {
 		_ = c.Close()
 	}()
 	require.NoError(t, err)
-	db, err := c.DB(context.Background(), "test")
+	const dbName = "test"
+	db, err := c.DB(context.Background(), dbName)
 	require.NoError(t, err)
 	table, err := db.Table("test", config)
 	require.NoError(t, err)
@@ -1904,7 +1905,7 @@ func Test_DB_DropStorage(t *testing.T) {
 	countRows(300)
 
 	level.Debug(logger).Log("msg", "dropping storage")
-	require.NoError(t, db.Close(WithClearStorage()))
+	require.NoError(t, c.DropDB(dbName))
 
 	// Open a new store against the dropped storage, and expect empty db
 	c, err = New(
Repository: polarsignals/frostdb
Pull request #596 (2023-11-27 10:45:09)

table: use pooled writers for compaction


Patch:
diff --git a/table.go b/table.go
index 2202e15..234ee2e 100644
--- a/table.go
+++ b/table.go
@@ -939,12 +939,8 @@ type parquetRowWriter struct {
 type parquetRowWriterOption func(p *parquetRowWriter)
 
 // rowWriter returns a new Parquet row writer with the given dynamic columns.
-func (t *TableBlock) rowWriter(writer io.Writer, dynCols map[string][]string, options ...parquetRowWriterOption) (*parquetRowWriter, error) {
-	w, err := t.table.schema.NewWriter(writer, dynCols, false)
-	if err != nil {
-		return nil, err
-	}
-
+// TODO(asubiotto): Can we delete this parquetRowWriter?
+func (t *TableBlock) rowWriter(w ParquetWriter, options ...parquetRowWriterOption) (*parquetRowWriter, error) {
 	buffSize := 256
 	if t.table.config.RowGroupSize > 0 {
 		buffSize = int(t.table.config.RowGroupSize)
@@ -1199,7 +1195,12 @@ func (t *Table) compactParts(w io.Writer, compact []*parts.Part) (int64, error)
 		return 0, err
 	}
 	err = func() error {
-		p, err := t.active.rowWriter(w, merged.DynamicColumns())
+		pw, err := t.schema.GetWriter(w, merged.DynamicColumns(), false)
+		if err != nil {
+			return err
+		}
+		defer t.schema.PutWriter(pw)
+		p, err := t.active.rowWriter(pw)
 		if err != nil {
 			return err
 		}


Test patch:
diff --git a/table_test.go b/table_test.go
index a2895f3..dc1e9c5 100644
--- a/table_test.go
+++ b/table_test.go
@@ -739,9 +739,12 @@ func Test_RowWriter(t *testing.T) {
 	defer c.Close()
 
 	b := &bytes.Buffer{}
-	rowWriter, err := table.ActiveBlock().rowWriter(b, map[string][]string{
+	pw, err := table.schema.GetWriter(b, map[string][]string{
 		"labels": {"node"},
-	})
+	}, false)
+	defer table.schema.PutWriter(pw)
+	require.NoError(t, err)
+	rowWriter, err := table.ActiveBlock().rowWriter(pw)
 	require.NoError(t, err)
 
 	// Write 17(8,9) rows, expect 3 row groups of 5 rows and 1 row group of 2 rows
Repository: polarsignals/frostdb
Pull request #678 (2024-01-09 16:02:59)

Remove explicit memory.Allocator for arrowutils.ReorderRecord
We already pass `context.Context` and use `compute`. Anyone who needs a custom Allocator can use `compute.WithAllocator`.

gernest: @thorfour I addressed your comment PTAL.
thorfour: Thanks!

Patch:
diff --git a/pqarrow/arrowutils/sort.go b/pqarrow/arrowutils/sort.go
index 1137a95..048c568 100644
--- a/pqarrow/arrowutils/sort.go
+++ b/pqarrow/arrowutils/sort.go
@@ -10,7 +10,6 @@ import (
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
 	"github.com/apache/arrow/go/v14/arrow/compute"
-	"github.com/apache/arrow/go/v14/arrow/memory"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/polarsignals/frostdb/pqarrow/builder"
@@ -75,8 +74,9 @@ func SortRecord(r arrow.Record, columns []SortingColumn) (*array.Int32, error) {
 
 // ReorderRecord uses indices to create a new record that is sorted according to
 // the indices array.
-func ReorderRecord(ctx context.Context, mem memory.Allocator, r arrow.Record, indices *array.Int32) (arrow.Record, error) {
-	ctx = compute.WithAllocator(ctx, mem)
+//
+// Use compute.WithAllocator to pass a custom memory.Allocator.
+func ReorderRecord(ctx context.Context, r arrow.Record, indices *array.Int32) (arrow.Record, error) {
 	// compute.Take doesn't support dictionaries. Use take on r when r does not have
 	// dictionary column.
 	var hasDictionary bool


Test patch:
diff --git a/pqarrow/arrowutils/sort_test.go b/pqarrow/arrowutils/sort_test.go
index aca8026..601fccd 100644
--- a/pqarrow/arrowutils/sort_test.go
+++ b/pqarrow/arrowutils/sort_test.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/apache/arrow/go/v14/arrow"
 	"github.com/apache/arrow/go/v14/arrow/array"
+	"github.com/apache/arrow/go/v14/arrow/compute"
 	"github.com/apache/arrow/go/v14/arrow/memory"
 	"github.com/stretchr/testify/require"
 
@@ -224,7 +225,8 @@ func TestReorderRecord(t *testing.T) {
 		indices := array.NewInt32Builder(mem)
 		indices.AppendValues([]int32{2, 1, 0}, nil)
 		by := indices.NewInt32Array()
-		result, err := arrowutils.ReorderRecord(context.Background(), mem, r, by)
+		result, err := arrowutils.ReorderRecord(
+			compute.WithAllocator(context.Background(), mem), r, by)
 		require.Nil(t, err)
 		defer result.Release()
 
@@ -256,7 +258,8 @@ func TestReorderRecord(t *testing.T) {
 		indices := array.NewInt32Builder(mem)
 		indices.AppendValues([]int32{2, 1, 0}, nil)
 		by := indices.NewInt32Array()
-		result, err := arrowutils.ReorderRecord(context.Background(), mem, r, by)
+		result, err := arrowutils.ReorderRecord(
+			compute.WithAllocator(context.Background(), mem), r, by)
 		require.Nil(t, err)
 		defer result.Release()
Repository: polarsignals/frostdb
Pull request #692 (2024-01-20 16:30:34)

Rename ReorderRecord to Take
The new name reflects what the function does, we are not ordering anything we are just taking specific rows.

gernest: I almost made a mistake to recommend duplicate implementation on ticket #672 because it was impossible to discover.

Patch:
diff --git a/pqarrow/arrowutils/sort.go b/pqarrow/arrowutils/sort.go
index 6364fec..e3a48e7 100644
--- a/pqarrow/arrowutils/sort.go
+++ b/pqarrow/arrowutils/sort.go
@@ -61,11 +61,11 @@ func SortRecord(r arrow.Record, columns []SortingColumn) (*array.Int32, error) {
 	return ms.indices.NewArray().(*array.Int32), nil
 }
 
-// ReorderRecord uses indices to create a new record that is sorted according to
-// the indices array.
+// Take uses indices which is an array of row index and returns a new record
+// that only contains rows specified in indices.
 //
 // Use compute.WithAllocator to pass a custom memory.Allocator.
-func ReorderRecord(ctx context.Context, r arrow.Record, indices *array.Int32) (arrow.Record, error) {
+func Take(ctx context.Context, r arrow.Record, indices *array.Int32) (arrow.Record, error) {
 	// compute.Take doesn't support dictionaries. Use take on r when r does not have
 	// dictionary column.
 	var hasDictionary bool


Test patch:
diff --git a/pqarrow/arrowutils/sort_test.go b/pqarrow/arrowutils/sort_test.go
index 601fccd..4fa8c4a 100644
--- a/pqarrow/arrowutils/sort_test.go
+++ b/pqarrow/arrowutils/sort_test.go
@@ -225,7 +225,7 @@ func TestReorderRecord(t *testing.T) {
 		indices := array.NewInt32Builder(mem)
 		indices.AppendValues([]int32{2, 1, 0}, nil)
 		by := indices.NewInt32Array()
-		result, err := arrowutils.ReorderRecord(
+		result, err := arrowutils.Take(
 			compute.WithAllocator(context.Background(), mem), r, by)
 		require.Nil(t, err)
 		defer result.Release()
@@ -258,7 +258,7 @@ func TestReorderRecord(t *testing.T) {
 		indices := array.NewInt32Builder(mem)
 		indices.AppendValues([]int32{2, 1, 0}, nil)
 		by := indices.NewInt32Array()
-		result, err := arrowutils.ReorderRecord(
+		result, err := arrowutils.Take(
 			compute.WithAllocator(context.Background(), mem), r, by)
 		require.Nil(t, err)
 		defer result.Release()
Repository: polarsignals/frostdb
Pull request #728 (2024-02-06 16:32:23)

db: notify txpool of watermark changes
Before a recent fix, we would rely on txn IDs to always be inserted into the TxPool to bubble up the watermark correctly. This commit adds a mechanism to notify the TxPool that the watermark has been changed, which is required in order to bubble up the watermark correctly.


Patch:
diff --git a/db.go b/db.go
index 7c93cc1..4bf531c 100644
--- a/db.go
+++ b/db.go
@@ -1266,6 +1266,7 @@ func (db *DB) begin() (uint64, uint64, func()) {
 		if mark := db.highWatermark.Load(); mark+1 == txn {
 			// This is the next consecutive transaction; increase the watermark.
 			db.highWatermark.Store(txn)
+			db.txPool.notifyWatermark()
 			return
 		}
 
diff --git a/tx_list.go b/tx_list.go
index 2590bee..0a09c7d 100644
--- a/tx_list.go
+++ b/tx_list.go
@@ -125,6 +125,15 @@ func (l *TxPool) insert(node, prev, next *TxNode) bool {
 	return success
 }
 
+// notifyWatermark notifies the TxPool that the watermark has been updated. This
+// triggers a sweep of the pool.
+func (l *TxPool) notifyWatermark() {
+	select {
+	case l.drain <- struct{}{}:
+	default:
+	}
+}
+
 func (l *TxPool) Iterate(iterate func(txn uint64) bool) {
 	for node := l.head.Load().next.Load(); node.tx != 0; node = getUnmarked(node) {
 		if isMarked(node) == nil && !iterate(node.tx) {


Test patch:
diff --git a/db_test.go b/db_test.go
index 6d0ad35..a03c58e 100644
--- a/db_test.go
+++ b/db_test.go
@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -2880,3 +2881,27 @@ func Test_DB_PersistentDiskCompaction_NonOverlappingCompaction(t *testing.T) {
 
 	validateRows(3)
 }
+
+func TestDBWatermarkBubbling(t *testing.T) {
+	c, err := New()
+	require.NoError(t, err)
+	defer c.Close()
+	db, err := c.DB(context.Background(), "test")
+	require.NoError(t, err)
+
+	const nTxns = 100
+	var wg sync.WaitGroup
+	wg.Add(nTxns)
+	for i := 0; i < nTxns; i++ {
+		_, _, commit := db.begin()
+		go func() {
+			defer wg.Done()
+			commit()
+		}()
+	}
+	wg.Wait()
+
+	require.Eventually(t, func() bool {
+		return db.HighWatermark() == nTxns
+	}, time.Second, 10*time.Millisecond)
+}
Repository: polarsignals/frostdb
Pull request #756 (2024-02-29 13:29:20)

physicalplan: short-circuit AndExpr if left side evaluates to false
Previously, filters run on records would run both the left and right expression on records. However, it is possible to short circuit the evaluation if the left expression evaluates to false on all rows in a record.


Patch:
diff --git a/query/physicalplan/filter.go b/query/physicalplan/filter.go
index 71a74a3..633af26 100644
--- a/query/physicalplan/filter.go
+++ b/query/physicalplan/filter.go
@@ -175,6 +175,10 @@ func (a *AndExpr) Eval(r arrow.Record) (*Bitmap, error) {
 		return nil, err
 	}
 
+	if left.IsEmpty() {
+		return left, nil
+	}
+
 	right, err := a.Right.Eval(r)
 	if err != nil {
 		return nil, err


Test patch:
diff --git a/query/physicalplan/filter_test.go b/query/physicalplan/filter_test.go
index d39d457..39978b7 100644
--- a/query/physicalplan/filter_test.go
+++ b/query/physicalplan/filter_test.go
@@ -3,6 +3,7 @@ package physicalplan
 import (
 	"testing"
 
+	"github.com/apache/arrow/go/v15/arrow"
 	"github.com/stretchr/testify/require"
 )
 
@@ -47,3 +48,34 @@ func Test_BuildIndexRanges(t *testing.T) {
 		})
 	}
 }
+
+type mockExpression struct {
+	evalFn func(arrow.Record) (*Bitmap, error)
+}
+
+var _ BooleanExpression = mockExpression{}
+
+func (e mockExpression) Eval(record arrow.Record) (*Bitmap, error) {
+	return e.evalFn(record)
+}
+
+func (e mockExpression) String() string {
+	return ""
+}
+
+func TestAndExprShortCircuits(t *testing.T) {
+	left := mockExpression{evalFn: func(_ arrow.Record) (*Bitmap, error) {
+		return NewBitmap(), nil
+	}}
+	right := mockExpression{evalFn: func(_ arrow.Record) (*Bitmap, error) {
+		t.Fatal("right should not be evaluated")
+		return nil, nil
+	}}
+	andExpr := AndExpr{
+		Left:  left,
+		Right: right,
+	}
+	result, err := andExpr.Eval(nil)
+	require.NoError(t, err)
+	require.True(t, result.IsEmpty())
+}