consider version in dsTxn Query

datastore/memory/txn.go

@@ -179,13 +179,17 @@ func (t *basicTxn) Query(ctx context.Context, q dsq.Query) (dsq.Results, error)
 	iter := t.ds.values.Iter()
 	iterOps := t.ops.Iter()
 	iterOpsHasValue := iterOps.Next()
+	dsVersion := t.getDSVersion()
 	// iterate over the underlying store and ensure that ops with keys smaller than or equal to
 	// the key of the underlying store are added with priority.
 	for iter.Next() {
 		// fast forward to last inserted version
 		item := iter.Item()
+		if item.version > dsVersion {
+			continue
+		}
 		for iter.Next() {
-			if item.key == iter.Item().key {
+			if item.key == iter.Item().key && iter.Item().version <= dsVersion {
 				item = iter.Item()
 				continue
 			}

datastore/memory/txn_test.go

@@ -661,7 +661,7 @@ func TestTxnQueryWithOnlyOneOperation(t *testing.T) {
 	tx, err := s.NewTransaction(ctx, false)
 	require.NoError(t, err)
 
-	err = s.Put(ctx, testKey4, testValue4)
+	err = tx.Put(ctx, testKey4, testValue4)
 	require.NoError(t, err)
 
 	results, err := tx.Query(ctx, dsq.Query{})

datastore/txn_test.go

@@ -486,8 +486,7 @@ func TestMemoryStoreTxn_TwoTransactionsWithQueryAndPut_ShouldOmmitNewPut(t *test
 	for r := range qResults.Next() {
 		docs = append(docs, r.Entry.Value)
 	}
-	// This is wrong. The new put should not be visible.
-	require.Equal(t, [][]byte{[]byte("value"), []byte("other-value")}, docs)
+	require.Equal(t, [][]byte{[]byte("value")}, docs)
 	txn1.Discard(ctx)
 }