style: Lint

Author: erezrokah

examples/simple_plugin/plugin/client.go

@@ -59,12 +59,12 @@ func (*Client) Write(context.Context, <-chan message.WriteMessage) error {
 	return nil
 }
 
-func (*Client) Read(context.Context, *schema.Table, chan<- arrow.Record) error {
+func (*Client) Read(context.Context, *schema.Table, chan<- arrow.RecordBatch) error {
 	// Not implemented, just used for testing destination packaging
 	return nil
 }
 
-func (*Client) Transform(_ context.Context, _ <-chan arrow.Record, _ chan<- arrow.Record) error {
+func (*Client) Transform(_ context.Context, _ <-chan arrow.RecordBatch, _ chan<- arrow.RecordBatch) error {
 	// Not implemented, just used for testing destination packaging
 	return nil
 }

internal/batch/slice.go

@@ -7,13 +7,13 @@ import (
 
 type (
 	SlicedRecord struct {
-		arrow.Record
+		arrow.RecordBatch
 		Bytes       int64 // we need this as the util.TotalRecordSize will report the full size even for the sliced record
 		bytesPerRow int64
 	}
 )
 
-func (s *SlicedRecord) split(limit *Cap) (add *SlicedRecord, toFlush []arrow.Record, rest *SlicedRecord) {
+func (s *SlicedRecord) split(limit *Cap) (add *SlicedRecord, toFlush []arrow.RecordBatch, rest *SlicedRecord) {
 	if s == nil {
 		return nil, nil, nil
 	}
@@ -23,13 +23,13 @@ func (s *SlicedRecord) split(limit *Cap) (add *SlicedRecord, toFlush []arrow.Rec
 		limit.add(add.Bytes, add.NumRows())
 	}
 
-	if s.Record == nil {
+	if s.RecordBatch == nil {
 		// all processed
 		return add, nil, nil
 	}
 
 	toFlush = s.getToFlush(limit)
-	if s.Record == nil {
+	if s.RecordBatch == nil {
 		// all processed
 		return add, toFlush, nil
 	}
@@ -56,21 +56,21 @@ func (s *SlicedRecord) getAdd(limit *Cap) *SlicedRecord {
 		// grab the whole record (either no limits or not overflowing)
 		res := *s
 		s.Bytes = 0
-		s.Record = nil
+		s.RecordBatch = nil
 		return &res
 	}
 
 	res := SlicedRecord{
-		Record:      s.NewSlice(0, rows),
+		RecordBatch: s.NewSlice(0, rows),
 		Bytes:       rows * s.bytesPerRow,
 		bytesPerRow: s.bytesPerRow,
 	}
-	s.Record = s.NewSlice(rows, s.NumRows())
+	s.RecordBatch = s.NewSlice(rows, s.NumRows())
 	s.Bytes -= res.Bytes
 	return &res
 }
 
-func (s *SlicedRecord) getToFlush(limit *Cap) []arrow.Record {
+func (s *SlicedRecord) getToFlush(limit *Cap) []arrow.RecordBatch {
 	rowsByBytes := limit.bytes.capPerN(s.bytesPerRow)
 	rows := limit.rows.cap()
 	switch {
@@ -93,41 +93,41 @@ func (s *SlicedRecord) getToFlush(limit *Cap) []arrow.Record {
 		return nil
 	}
 
-	flush := make([]arrow.Record, 0, s.NumRows()/rows)
+	flush := make([]arrow.RecordBatch, 0, s.NumRows()/rows)
 	offset := int64(0)
 	for offset+rows <= s.NumRows() {
 		flush = append(flush, s.NewSlice(offset, offset+rows))
 		offset += rows
 	}
 	if offset == s.NumRows() {
 		// we processed everything for flush
-		s.Record = nil
+		s.RecordBatch = nil
 		s.Bytes = 0
 		return flush
 	}
 
 	// set record to the remainder
-	s.Record = s.NewSlice(offset, s.NumRows())
+	s.RecordBatch = s.NewSlice(offset, s.NumRows())
 	s.Bytes = s.NumRows() * s.bytesPerRow
 
 	return flush
 }
 
-func (s *SlicedRecord) slice() []arrow.Record {
-	res := make([]arrow.Record, s.NumRows())
+func (s *SlicedRecord) slice() []arrow.RecordBatch {
+	res := make([]arrow.RecordBatch, s.NumRows())
 	for i := int64(0); i < s.NumRows(); i++ {
 		res[i] = s.NewSlice(i, i+1)
 	}
 	return res
 }
 
-func newSlicedRecord(r arrow.Record) *SlicedRecord {
+func newSlicedRecord(r arrow.RecordBatch) *SlicedRecord {
 	if r.NumRows() == 0 {
 		return nil
 	}
 	res := SlicedRecord{
-		Record: r,
-		Bytes:  util.TotalRecordSize(r),
+		RecordBatch: r,
+		Bytes:       util.TotalRecordSize(r),
 	}
 	res.bytesPerRow = res.Bytes / r.NumRows()
 	return &res
@@ -136,10 +136,10 @@ func newSlicedRecord(r arrow.Record) *SlicedRecord {
 // SliceRecord will return the SlicedRecord you can add to the batch given the restrictions provided (if any).
 // The meaning of the returned values:
 // - `add` is good to be added to the current batch that the caller is assembling
-// - `flush` represents sliced arrow.Record that needs own batch to be flushed
+// - `flush` represents sliced arrow.RecordBatch that needs own batch to be flushed
 // - `remaining` represents the overflow of the batch after `add` & `flush` are processed
 // Note that the `limit` provided will not be updated.
-func SliceRecord(r arrow.Record, limit *Cap) (add *SlicedRecord, flush []arrow.Record, remaining *SlicedRecord) {
+func SliceRecord(r arrow.RecordBatch, limit *Cap) (add *SlicedRecord, flush []arrow.RecordBatch, remaining *SlicedRecord) {
 	l := *limit // copy value
 	return newSlicedRecord(r).split(&l)
 }

internal/clients/state/v3/state.go

@@ -163,7 +163,7 @@ func (c *Client) Flush(ctx context.Context) error {
 			version.Append(val.version)
 		}
 	}
-	rec := bldr.NewRecord()
+	rec := bldr.NewRecordBatch()
 	recordBytes, err := pb.RecordToBytes(rec)
 	if err != nil {
 		return err

internal/memdb/memdb.go

@@ -16,7 +16,7 @@ import (
 
 // client is mostly used for testing the destination plugin.
 type client struct {
-	memoryDB      map[string][]arrow.Record
+	memoryDB      map[string][]arrow.RecordBatch
 	tables        map[string]*schema.Table
 	memoryDBLock  sync.RWMutex
 	errOnWrite    bool
@@ -42,7 +42,7 @@ func WithBlockingWrite() Option {
 
 func GetNewClient(options ...Option) plugin.NewClientFunc {
 	c := &client{
-		memoryDB:     make(map[string][]arrow.Record),
+		memoryDB:     make(map[string][]arrow.RecordBatch),
 		memoryDBLock: sync.RWMutex{},
 		tables: map[string]*schema.Table{
 			"table1": {
@@ -112,13 +112,13 @@ func NewMemDBClientErrOnNew(context.Context, zerolog.Logger, []byte, plugin.NewC
 	return nil, errors.New("newTestDestinationMemDBClientErrOnNew")
 }
 
-func (c *client) overwrite(table *schema.Table, record arrow.Record) {
+func (c *client) overwrite(table *schema.Table, record arrow.RecordBatch) {
 	for i := int64(0); i < record.NumRows(); i++ {
 		c.overwriteRow(table, record.NewSlice(i, i+1))
 	}
 }
 
-func (c *client) overwriteRow(table *schema.Table, data arrow.Record) {
+func (c *client) overwriteRow(table *schema.Table, data arrow.RecordBatch) {
 	tableName := table.Name
 	pksIndex := table.PrimaryKeysIndexes()
 	if len(pksIndex) == 0 {
@@ -152,7 +152,7 @@ func (*client) GetSpec() any {
 	return &Spec{}
 }
 
-func (c *client) Read(_ context.Context, table *schema.Table, res chan<- arrow.Record) error {
+func (c *client) Read(_ context.Context, table *schema.Table, res chan<- arrow.RecordBatch) error {
 	c.memoryDBLock.RLock()
 	defer c.memoryDBLock.RUnlock()
 
@@ -196,7 +196,7 @@ func (c *client) migrate(_ context.Context, table *schema.Table) {
 	tableName := table.Name
 	memTable := c.memoryDB[tableName]
 	if memTable == nil {
-		c.memoryDB[tableName] = make([]arrow.Record, 0)
+		c.memoryDB[tableName] = make([]arrow.RecordBatch, 0)
 		c.tables[tableName] = table
 		return
 	}
@@ -206,7 +206,7 @@ func (c *client) migrate(_ context.Context, table *schema.Table) {
 	if changes == nil {
 		return
 	}
-	c.memoryDB[tableName] = make([]arrow.Record, 0)
+	c.memoryDB[tableName] = make([]arrow.RecordBatch, 0)
 	c.tables[tableName] = table
 }
 
@@ -253,7 +253,7 @@ func (c *client) Close(context.Context) error {
 }
 
 func (c *client) deleteStale(_ context.Context, msg *message.WriteDeleteStale) {
-	var filteredTable []arrow.Record
+	var filteredTable []arrow.RecordBatch
 	tableName := msg.TableName
 	for i, row := range c.memoryDB[tableName] {
 		sc := row.Schema()
@@ -280,7 +280,7 @@ func (c *client) deleteStale(_ context.Context, msg *message.WriteDeleteStale) {
 }
 
 func (c *client) deleteRecord(_ context.Context, msg *message.WriteDeleteRecord) {
-	var filteredTable []arrow.Record
+	var filteredTable []arrow.RecordBatch
 	tableName := msg.TableName
 	for i, row := range c.memoryDB[tableName] {
 		isMatch := true
@@ -308,15 +308,15 @@ func (c *client) deleteRecord(_ context.Context, msg *message.WriteDeleteRecord)
 	c.memoryDB[tableName] = filteredTable
 }
 
-func (*client) Transform(_ context.Context, _ <-chan arrow.Record, _ chan<- arrow.Record) error {
+func (*client) Transform(_ context.Context, _ <-chan arrow.RecordBatch, _ chan<- arrow.RecordBatch) error {
 	return nil
 }
 
 func (*client) TransformSchema(_ context.Context, _ *arrow.Schema) (*arrow.Schema, error) {
 	return nil, nil
 }
 
-func evaluatePredicate(pred message.Predicate, record arrow.Record) bool {
+func evaluatePredicate(pred message.Predicate, record arrow.RecordBatch) bool {
 	sc := record.Schema()
 	indices := sc.FieldIndices(pred.Column)
 	if len(indices) == 0 {

internal/memdb/memdb_test.go

@@ -62,7 +62,7 @@ func TestOnWriteError(t *testing.T) {
 // 	sourceSpec := pbPlugin.Spec{
 // 		Name: sourceName,
 // 	}
-// 	ch := make(chan arrow.Record, 1)
+// 	ch := make(chan arrow.RecordBatch, 1)
 // 	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
 // 	opts := schema.GenTestDataOptions{
 // 		SourceName: "test",

internal/pk/pk.go

@@ -7,7 +7,7 @@ import (
 	"github.com/cloudquery/plugin-sdk/v4/schema"
 )
 
-func String(resource arrow.Record) string {
+func String(resource arrow.RecordBatch) string {
 	sc := resource.Schema()
 	table, err := schema.NewTableFromArrowSchema(sc)
 	if err != nil {

internal/reversertransformer/reversertransformer.go

@@ -40,7 +40,7 @@ func (*client) Close(context.Context) error {
 	return nil
 }
 
-func (c *client) Transform(ctx context.Context, recvRecords <-chan arrow.Record, sendRecords chan<- arrow.Record) error {
+func (c *client) Transform(ctx context.Context, recvRecords <-chan arrow.RecordBatch, sendRecords chan<- arrow.RecordBatch) error {
 	for {
 		select {
 		case record, ok := <-recvRecords:
@@ -62,7 +62,7 @@ func (*client) TransformSchema(_ context.Context, old *arrow.Schema) (*arrow.Sch
 	return old, nil
 }
 
-func (*client) reverseStrings(record arrow.Record) (arrow.Record, error) {
+func (*client) reverseStrings(record arrow.RecordBatch) (arrow.RecordBatch, error) {
 	for i, column := range record.Columns() {
 		if column.DataType().ID() != arrow.STRING {
 			continue

internal/reversertransformer/reversertransformer_test.go

@@ -55,13 +55,13 @@ func makeRequestFromString(s string) *pb.Transform_Request {
 	return &pb.Transform_Request{Record: bs}
 }
 
-func makeRecordFromString(s string) arrow.Record {
+func makeRecordFromString(s string) arrow.RecordBatch {
 	str := array.NewStringBuilder(memory.DefaultAllocator)
 	str.AppendString(s)
 	arr := str.NewStringArray()
 	schema := arrow.NewSchema([]arrow.Field{{Name: "col1", Type: arrow.BinaryTypes.String}}, nil)
 
-	return array.NewRecord(schema, []arrow.Array{arr}, 1)
+	return array.NewRecordBatch(schema, []arrow.Array{arr}, 1)
 }
 
 type mockTransformServer struct {

internal/servers/destination/v0/schemav2tov3.go

@@ -96,11 +96,11 @@ func TypeV2ToV3(dataType schemav2.ValueType) arrow.DataType {
 	}
 }
 
-func CQTypesOneToRecord(mem memory.Allocator, c schemav2.CQTypes, arrowSchema *arrow.Schema) arrow.Record {
+func CQTypesOneToRecord(mem memory.Allocator, c schemav2.CQTypes, arrowSchema *arrow.Schema) arrow.RecordBatch {
 	return CQTypesToRecord(mem, []schemav2.CQTypes{c}, arrowSchema)
 }
 
-func CQTypesToRecord(mem memory.Allocator, c []schemav2.CQTypes, arrowSchema *arrow.Schema) arrow.Record {
+func CQTypesToRecord(mem memory.Allocator, c []schemav2.CQTypes, arrowSchema *arrow.Schema) arrow.RecordBatch {
 	bldr := array.NewRecordBuilder(mem, arrowSchema)
 	fields := bldr.Fields()
 	for i := range fields {
@@ -242,5 +242,5 @@ func CQTypesToRecord(mem memory.Allocator, c []schemav2.CQTypes, arrowSchema *ar
 		}
 	}
 
-	return bldr.NewRecord()
+	return bldr.NewRecordBatch()
 }

internal/servers/destination/v1/destination_test.go

@@ -121,7 +121,7 @@ func TestPluginSync(t *testing.T) {
 	sc := table.ToArrowSchema()
 	bldr := array.NewRecordBuilder(memory.DefaultAllocator, sc)
 	bldr.Field(0).(*array.StringBuilder).Append("test")
-	record := bldr.NewRecord()
+	record := bldr.NewRecordBatch()
 	recordBytes, err := pbSource.RecordToBytes(record)
 	if err != nil {
 		t.Fatal(err)