1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
package parquet
import (
"path/filepath"
"testing"
)
var benchmarkRecordSink Record
func BenchmarkWriterThroughput(b *testing.B) {
rows := benchmarkRecords(256)
dir := b.TempDir()
writer, err := NewWriter(filepath.Join(dir, "writer-throughput.parquet"), WriterConfig{}, FileMetadata{Mode: "bench"})
if err != nil {
b.Fatalf("NewWriter() error = %v", err)
}
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
if err := writer.WriteRows(rows); err != nil {
b.Fatalf("WriteRows() error = %v", err)
}
benchmarkRecordSink = rows[i%len(rows)]
}
b.StopTimer()
if err := writer.Close(); err != nil {
b.Fatalf("Close() error = %v", err)
}
}
func BenchmarkRecorderQueueHandoff(b *testing.B) {
row := testStreamRow(1, "read", false)
session := newRecordingSession(1)
recorder := &Recorder{
active: session,
status: Status{Active: true},
}
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
row.Seq = uint64(i + 1)
if err := recorder.Record(row, 0); err != nil {
b.Fatalf("Record() error = %v", err)
}
select {
case <-session.queue:
default:
b.Fatal("expected queued record request")
}
}
b.StopTimer()
session.stop(nil)
recorder.finishSession(session, nil)
}
func benchmarkRecords(n int) []Record {
rows := make([]Record, 0, n)
for i := 0; i < n; i++ {
row := testStreamRow(uint64(i+1), "read", i%7 == 0)
rows = append(rows, RecordFromStream(row, uint64(i%4)))
}
return rows
}
|
