diff --git a/csv/write_read_test.go b/csv/write_read_test.go
index a4b3c1aa..af90cad9 100644
--- a/csv/write_read_test.go
+++ b/csv/write_read_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/cloudquery/plugin-sdk/v4/plugin"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWriteRead(t *testing.T) {
@@ -40,7 +41,7 @@ func TestWriteRead(t *testing.T) {
 				StableTime: time.Date(2021, 1, 2, 0, 0, 0, 0, time.UTC),
 			}
 			tg := schema.NewTestDataGenerator()
-			records := tg.Generate(table, opts)
+			record := tg.Generate(table, opts)
 
 			cl, err := NewClient(tc.options...)
 			if err != nil {
@@ -51,7 +52,7 @@ func TestWriteRead(t *testing.T) {
 			writer := bufio.NewWriter(&b)
 			reader := bufio.NewReader(&b)
 
-			if err := types.WriteAll(cl, writer, table, records); err != nil {
+			if err := types.WriteAll(cl, writer, table, []arrow.Record{record}); err != nil {
 				t.Fatal(err)
 			}
 			writer.Flush()
@@ -74,19 +75,13 @@ func TestWriteRead(t *testing.T) {
 				readErr = cl.Read(byteReader, table, ch)
 				close(ch)
 			}()
-			totalCount := 0
+			received := make([]arrow.Record, 0, tc.outputCount)
 			for got := range ch {
-				if diff := plugin.RecordDiff(records[totalCount], got); diff != "" {
-					t.Errorf("got diff: %s", diff)
-				}
-				totalCount++
-			}
-			if readErr != nil {
-				t.Fatal(readErr)
-			}
-			if totalCount != tc.outputCount {
-				t.Errorf("got %d row(s), want %d", totalCount, tc.outputCount)
+				received = append(received, got)
 			}
+			require.Empty(t, plugin.RecordsDiff(table.ToArrowSchema(), []arrow.Record{record}, received))
+			require.NoError(t, readErr)
+			require.Equalf(t, tc.outputCount, len(received), "got %d row(s), want %d", len(received), tc.outputCount)
 		})
 	}
 }
@@ -101,7 +96,7 @@ func BenchmarkWrite(b *testing.B) {
 		MaxRows:    1000,
 	}
 	tg := schema.NewTestDataGenerator()
-	records := tg.Generate(table, opts)
+	record := tg.Generate(table, opts)
 
 	cl, err := NewClient()
 	if err != nil {
@@ -111,7 +106,7 @@ func BenchmarkWrite(b *testing.B) {
 	writer := bufio.NewWriter(&buf)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		if err := types.WriteAll(cl, writer, table, records); err != nil {
+		if err := types.WriteAll(cl, writer, table, []arrow.Record{record}); err != nil {
 			b.Fatal(err)
 		}
 
diff --git a/go.mod b/go.mod
index 8efc3eeb..c06d9b4d 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.19
 require (
 	github.com/apache/arrow/go/v13 v13.0.0-20230731205701-112f94971882
 	github.com/bradleyjkemp/cupaloy/v2 v2.8.0
-	github.com/cloudquery/plugin-sdk/v4 v4.5.1
+	github.com/cloudquery/plugin-sdk/v4 v4.5.2
 	github.com/goccy/go-json v0.10.2
 	github.com/google/uuid v1.3.0
 	github.com/stretchr/testify v1.8.4
diff --git a/go.sum b/go.sum
index b2ca975e..b9a40938 100644
--- a/go.sum
+++ b/go.sum
@@ -8,8 +8,8 @@ github.com/bradleyjkemp/cupaloy/v2 v2.8.0 h1:any4BmKE+jGIaMpnU8YgH/I2LPiLBufr6oM
 github.com/bradleyjkemp/cupaloy/v2 v2.8.0/go.mod h1:bm7JXdkRd4BHJk9HpwqAI8BoAY1lps46Enkdqw6aRX0=
 github.com/cloudquery/arrow/go/v13 v13.0.0-20230821001142-5b83d4f91697 h1:D6qQZxiBMSz7YiB3nokGxg7BF+46UMEk2lPJmcifJg0=
 github.com/cloudquery/arrow/go/v13 v13.0.0-20230821001142-5b83d4f91697/go.mod h1:W69eByFNO0ZR30q1/7Sr9d83zcVZmF2MiP3fFYAWJOc=
-github.com/cloudquery/plugin-sdk/v4 v4.5.1 h1:V6EfHkW6kyYVMSX2S/Fn1b+62bU1jegIuv16Koo2f98=
-github.com/cloudquery/plugin-sdk/v4 v4.5.1/go.mod h1:WNTjNp8CPDDSpJHOSaho+MUzntKO6czG6E+yvVe++aA=
+github.com/cloudquery/plugin-sdk/v4 v4.5.2 h1:evRxri2pw/+Iohu7RNXcjd1PZiZkaXUP0T8CgtgtXv4=
+github.com/cloudquery/plugin-sdk/v4 v4.5.2/go.mod h1:/2/9y6iODo8jOT0mrAp8K1eogr7+228cy53jK8aUCjY=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
diff --git a/json/write_read_test.go b/json/write_read_test.go
index 29d491fa..e4e51b17 100644
--- a/json/write_read_test.go
+++ b/json/write_read_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/cloudquery/plugin-sdk/v4/plugin"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWriteRead(t *testing.T) {
@@ -27,7 +28,7 @@ func TestWriteRead(t *testing.T) {
 		StableTime: time.Date(2021, 1, 2, 0, 0, 0, 0, time.UTC),
 	}
 	tg := schema.NewTestDataGenerator()
-	records := tg.Generate(table, opts)
+	record := tg.Generate(table, opts)
 
 	cl, err := NewClient()
 	if err != nil {
@@ -38,7 +39,7 @@ func TestWriteRead(t *testing.T) {
 	writer := bufio.NewWriter(&b)
 	reader := bufio.NewReader(&b)
 
-	if err := types.WriteAll(cl, writer, table, records); err != nil {
+	if err := types.WriteAll(cl, writer, table, []arrow.Record{record}); err != nil {
 		t.Fatal(err)
 	}
 	writer.Flush()
@@ -61,19 +62,13 @@ func TestWriteRead(t *testing.T) {
 		readErr = cl.Read(byteReader, table, ch)
 		close(ch)
 	}()
-	totalCount := 0
+	received := make([]arrow.Record, 0, 2)
 	for got := range ch {
-		if diff := plugin.RecordDiff(records[totalCount], got); diff != "" {
-			t.Fatalf("got diff: %s", diff)
-		}
-		totalCount++
-	}
-	if readErr != nil {
-		t.Fatal(readErr)
-	}
-	if totalCount != 2 {
-		t.Fatalf("expected 2 rows, got %d", totalCount)
+		received = append(received, got)
 	}
+	require.Empty(t, plugin.RecordsDiff(table.ToArrowSchema(), []arrow.Record{record}, received))
+	require.NoError(t, readErr)
+	require.Equalf(t, 2, len(received), "got %d row(s), want %d", len(received), 2)
 }
 
 func BenchmarkWrite(b *testing.B) {
@@ -86,7 +81,7 @@ func BenchmarkWrite(b *testing.B) {
 		MaxRows:    1000,
 	}
 	tg := schema.NewTestDataGenerator()
-	records := tg.Generate(table, opts)
+	record := tg.Generate(table, opts)
 
 	cl, err := NewClient()
 	if err != nil {
@@ -96,7 +91,7 @@ func BenchmarkWrite(b *testing.B) {
 	writer := bufio.NewWriter(&buf)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		if err := types.WriteAll(cl, writer, table, records); err != nil {
+		if err := types.WriteAll(cl, writer, table, []arrow.Record{record}); err != nil {
 			b.Fatal(err)
 		}
 		err = writer.Flush()
diff --git a/parquet/write_read_test.go b/parquet/write_read_test.go
index 5d643ea9..e9ffb9af 100644
--- a/parquet/write_read_test.go
+++ b/parquet/write_read_test.go
@@ -8,10 +8,10 @@ import (
 	"time"
 
 	"github.com/apache/arrow/go/v13/arrow"
-	"github.com/apache/arrow/go/v13/arrow/array"
 	"github.com/cloudquery/filetypes/v4/types"
 	"github.com/cloudquery/plugin-sdk/v4/plugin"
 	"github.com/cloudquery/plugin-sdk/v4/schema"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWriteRead(t *testing.T) {
@@ -25,7 +25,7 @@ func TestWriteRead(t *testing.T) {
 		MaxRows:    2,
 	}
 	tg := schema.NewTestDataGenerator()
-	records := tg.Generate(table, opts)
+	record := tg.Generate(table, opts)
 
 	writer := bufio.NewWriter(&b)
 	reader := bufio.NewReader(&b)
@@ -34,7 +34,7 @@ func TestWriteRead(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err := types.WriteAll(cl, writer, table, records); err != nil {
+	if err := types.WriteAll(cl, writer, table, []arrow.Record{record}); err != nil {
 		t.Fatal(err)
 	}
 	err = writer.Flush()
@@ -53,20 +53,13 @@ func TestWriteRead(t *testing.T) {
 		readErr = cl.Read(byteReader, table, ch)
 		close(ch)
 	}()
-	totalCount := 0
+	received := make([]arrow.Record, 0, 2)
 	for got := range ch {
-		curr := records[totalCount]
-		if !array.RecordApproxEqual(curr, got) {
-			t.Fatalf("got diff (record %d): %s\n", totalCount, plugin.RecordDiff(records[totalCount], got))
-		}
-		totalCount++
-	}
-	if readErr != nil {
-		t.Fatal(readErr)
-	}
-	if totalCount != 2 {
-		t.Fatalf("expected 2 rows, got %d", totalCount)
+		received = append(received, got)
 	}
+	require.Empty(t, plugin.RecordsDiff(table.ToArrowSchema(), []arrow.Record{record}, received))
+	require.NoError(t, readErr)
+	require.Equalf(t, 2, len(received), "got %d row(s), want %d", len(received), 2)
 }
 
 func BenchmarkWrite(b *testing.B) {
@@ -79,7 +72,7 @@ func BenchmarkWrite(b *testing.B) {
 		MaxRows:    b.N,
 	}
 	tg := schema.NewTestDataGenerator()
-	records := tg.Generate(table, opts)
+	record := tg.Generate(table, opts)
 
 	cl, err := NewClient()
 	if err != nil {
@@ -89,7 +82,7 @@ func BenchmarkWrite(b *testing.B) {
 	writer := bufio.NewWriter(&buf)
 	b.ResetTimer()
 
-	if err := types.WriteAll(cl, writer, table, records); err != nil {
+	if err := types.WriteAll(cl, writer, table, []arrow.Record{record}); err != nil {
 		b.Fatal(err)
 	}
 	err = writer.Flush()