Repository: talariadb/talaria (language: python)

Issue:
Refactor Sink Writer to accept blocks inplace of bytes



Discussion / hints:
Looks like we need to fix some tests
Tests are fixed.
> Looks like we need to fix some tests

Fixed
There's a few conflicts @tardunge 
> There's a few conflicts @tardunge

will fix.
@tardunge please address comments in a separate PR(s)
> 

Sure.
This MR introduced an issue that the Azure writer would not initiate. @tardunge  

> This MR introduced an issue that the Azure writer would not initiate. @tardunge

Can you create an issue and post the error or behaviour you are encountering and also the expected behaviour?
@tardunge hi, here it is https://github.com/talariadb/talaria/issues/103

Fix patch:
diff --git a/.gitignore b/.gitignore
index f7f8cbbf..725b2d5c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,4 +15,5 @@ logs/
 *.venv
 .vscode/
 talaria
-*.so
\ No newline at end of file
+*.so
+local/
diff --git a/internal/encoding/block/block.go b/internal/encoding/block/block.go
index bc985a7d..95b447c1 100644
--- a/internal/encoding/block/block.go
+++ b/internal/encoding/block/block.go
@@ -8,12 +8,10 @@ import (
 	"errors"
 	"fmt"
 
-	eorc "github.com/crphang/orc"
 	"github.com/golang/snappy"
 	"github.com/kelindar/binary"
 	"github.com/kelindar/binary/nocopy"
 	"github.com/kelindar/talaria/internal/column"
-	"github.com/kelindar/talaria/internal/encoding/orc"
 	"github.com/kelindar/talaria/internal/encoding/typeof"
 	"github.com/kelindar/talaria/internal/presto"
 )
@@ -35,30 +33,6 @@ type Block struct {
 	schema  typeof.Schema  `binary:"-"` // The cached schema of the block
 }
 
-// Create a base block for testing purpose
-func Base() ([]Block, error) {
-	schema := typeof.Schema{
-		"col0": typeof.String,
-		"col1": typeof.Int64,
-		"col2": typeof.Float64,
-	}
-	orcSchema, _ := orc.SchemaFor(schema)
-
-	orcBuffer1 := &bytes.Buffer{}
-	writer, _ := eorc.NewWriter(orcBuffer1,
-		eorc.SetSchema(orcSchema))
-	_ = writer.Write("eventName", 1, 1.0)
-	_ = writer.Close()
-
-	apply := Transform(nil)
-
-	block, err := FromOrcBy(orcBuffer1.Bytes(), "col0", nil, apply)
-	if err != nil {
-		return nil, err
-	}
-	return block, nil
-}
-
 // Read decodes the block and selects the columns
 func Read(buffer []byte, desiredSchema typeof.Schema) (column.Columns, error) {
 	block, err := FromBuffer(buffer)
@@ -400,3 +374,26 @@ func decodeValue(kind typeof.Type, b []byte) (presto.Column, error) {
 }
 
 var emptyBlock = presto.PrestoThriftBlock{}
+
+// FromBlockBy creates and returns a list of new block.Row for a block.
+func FromBlockBy(blk Block, schema typeof.Schema) ([]Row, error) {
+	cols, err := blk.Select(schema)
+	if err != nil {
+		return nil, err
+	}
+	rowCount := cols.Any().Count()
+	rows := make([]Row, rowCount)
+	for i := 0; i < rowCount; i++ {
+		row := NewRow(schema, len(schema))
+		for name := range schema {
+			col := cols[name]
+			val := col.At(i)
+			if val == nil {
+				continue
+			}
+			row.Set(name, val)
+		}
+		rows[i] = row
+	}
+	return rows, nil
+}
diff --git a/internal/encoding/block/from_block.go b/internal/encoding/block/from_block.go
deleted file mode 100644
index 7d865c2d..00000000
--- a/internal/encoding/block/from_block.go
+++ /dev/null
@@ -1,31 +0,0 @@
-// Copyright 2019-2020 Grabtaxi Holdings PTE LTE (GRAB), All rights reserved.
-// Use of this source code is governed by an MIT-style license that can be found in the LICENSE file
-
-package block
-
-import (
-	"github.com/kelindar/talaria/internal/encoding/typeof"
-)
-
-// FromBlockBy creates and returns a list of new block.Row for a block.
-func FromBlockBy(blk Block, schema typeof.Schema) ([]Row, error) {
-	cols, err := blk.Select(schema)
-	if err != nil {
-		return nil, err
-	}
-	rowCount := cols.Any().Count()
-	rows := make([]Row, rowCount)
-	for i := 0; i < rowCount; i++ {
-		row := NewRow(schema, len(schema))
-		for name := range schema {
-			col := cols[name]
-			val := col.At(i)
-			if val == nil {
-				continue
-			}
-			row.Set(name, val)
-		}
-		rows[i] = row
-	}
-	return rows, nil
-}
diff --git a/internal/storage/flush/flush.go b/internal/storage/flush/flush.go
index 0174b6c7..0c1a6c1d 100644
--- a/internal/storage/flush/flush.go
+++ b/internal/storage/flush/flush.go
@@ -42,12 +42,6 @@ func (s *Flusher) WriteBlock(blocks []block.Block, schema typeof.Schema) error {
 		return nil
 	}
 
-	// Merge the blocks based on the specified merging function
-	// buffer, err := s.merge(blocks, schema)
-	// if err != nil {
-	//     return err
-	// }
-
 	// Generate the file name and write the data to the underlying writer
 	return s.writer.Write(s.generateFileName(blocks[0]), blocks)
 }


Test patch:
diff --git a/internal/encoding/block/block_test.go b/internal/encoding/block/block_test.go
index f1afb9cd..abe30f97 100644
--- a/internal/encoding/block/block_test.go
+++ b/internal/encoding/block/block_test.go
@@ -11,6 +11,25 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func setup() (Block, typeof.Schema) {
+
+	const testFile = "../../../test/test4.csv"
+	o, _ := ioutil.ReadFile(testFile)
+	schema := &typeof.Schema{
+		"raisedCurrency": typeof.String,
+		"raisedAmt":      typeof.Float64,
+	}
+	apply := Transform(schema)
+	b, _ := FromCSVBy(o, "raisedCurrency", &typeof.Schema{
+		"raisedCurrency": typeof.String,
+		"raisedAmt":      typeof.Float64,
+	}, apply)
+	if len(b) > 0 {
+		return b[0], *schema
+	}
+	return Block{}, *schema
+}
+
 func TestBlock_Types(t *testing.T) {
 	o, err := ioutil.ReadFile(smallFile)
 	assert.NotEmpty(t, o)
@@ -160,3 +179,20 @@ func noerror(err error) {
 		panic(err)
 	}
 }
+
+func TestFromBlock(t *testing.T) {
+	blk, schema := setup()
+	rows, err := FromBlockBy(blk, schema)
+	assert.NoError(t, err)
+	cols, err := blk.Select(schema)
+	assert.NoError(t, err)
+	rowCount := cols.Any().Count()
+	// verify row count.
+	assert.Equal(t, rowCount, len(rows))
+	for _, row := range rows {
+		// verify values
+		assert.Contains(t, []string{"EUR", "CAD", "USD"}, row.Values["raisedCurrency"])
+		// verify type
+		assert.Equal(t, typeof.String, row.Schema["raisedCurrency"])
+	}
+}
diff --git a/internal/encoding/block/from_block_test.go b/internal/encoding/block/from_block_test.go
deleted file mode 100644
index 2f444ddf..00000000
--- a/internal/encoding/block/from_block_test.go
+++ /dev/null
@@ -1,48 +0,0 @@
-// Copyright (c) Roman Atachiants and contributors. All rights reserved.
-// Licensed under the MIT license. See LICENSE file in the project root for details.
-
-package block
-
-import (
-	"io/ioutil"
-	"testing"
-
-	"github.com/kelindar/talaria/internal/encoding/typeof"
-	"github.com/stretchr/testify/assert"
-)
-
-func setup() (Block, typeof.Schema) {
-
-	const testFile = "../../../test/test4.csv"
-	o, _ := ioutil.ReadFile(testFile)
-	schema := &typeof.Schema{
-		"raisedCurrency": typeof.String,
-		"raisedAmt":      typeof.Float64,
-	}
-	apply := Transform(schema)
-	b, _ := FromCSVBy(o, "raisedCurrency", &typeof.Schema{
-		"raisedCurrency": typeof.String,
-		"raisedAmt":      typeof.Float64,
-	}, apply)
-	if len(b) > 0 {
-		return b[0], *schema
-	}
-	return Block{}, *schema
-}
-
-func TestFromBlock(t *testing.T) {
-	blk, schema := setup()
-	rows, err := FromBlockBy(blk, schema)
-	assert.NoError(t, err)
-	cols, err := blk.Select(schema)
-	assert.NoError(t, err)
-	rowCount := cols.Any().Count()
-	// verify row count.
-	assert.Equal(t, rowCount, len(rows))
-	for _, row := range rows {
-		// verify values
-		assert.Contains(t, []string{"EUR", "CAD", "USD"}, row.Values["raisedCurrency"])
-		// verify type
-		assert.Equal(t, typeof.String, row.Schema["raisedCurrency"])
-	}
-}
diff --git a/internal/storage/writer/azure/azure_test.go b/internal/storage/writer/azure/azure_test.go
index 47f17c1b..94f29b79 100644
--- a/internal/storage/writer/azure/azure_test.go
+++ b/internal/storage/writer/azure/azure_test.go
@@ -1,17 +1,45 @@
 package azure
 
 import (
+	"bytes"
 	"os"
 	"strings"
 	"testing"
 
+	eorc "github.com/crphang/orc"
 	"github.com/kelindar/talaria/internal/encoding/block"
+	"github.com/kelindar/talaria/internal/encoding/orc"
+	"github.com/kelindar/talaria/internal/encoding/typeof"
 	"github.com/kelindar/talaria/internal/monitor"
 	"github.com/kelindar/talaria/internal/monitor/logging"
 	"github.com/kelindar/talaria/internal/monitor/statsd"
 	"github.com/stretchr/testify/assert"
 )
 
+// Create a base block for testing purpose
+func blockBase() ([]block.Block, error) {
+	schema := typeof.Schema{
+		"col0": typeof.String,
+		"col1": typeof.Int64,
+		"col2": typeof.Float64,
+	}
+	orcSchema, _ := orc.SchemaFor(schema)
+
+	orcBuffer1 := &bytes.Buffer{}
+	writer, _ := eorc.NewWriter(orcBuffer1,
+		eorc.SetSchema(orcSchema))
+	_ = writer.Write("eventName", 1, 1.0)
+	_ = writer.Close()
+
+	apply := block.Transform(nil)
+
+	block, err := block.FromOrcBy(orcBuffer1.Bytes(), "col0", nil, apply)
+	if err != nil {
+		return nil, err
+	}
+	return block, nil
+}
+
 func TestWriter(t *testing.T) {
 	os.Setenv("AZURE_STORAGE_ACCOUNT", "golangrocksonazure")
 	os.Setenv("AZURE_STORAGE_ACCESS_KEY", "YmFy")
@@ -20,7 +48,7 @@ func TestWriter(t *testing.T) {
 	assert.NotNil(t, c)
 	assert.NoError(t, err)
 
-	b, err := block.Base()
+	b, err := blockBase()
 	assert.Nil(t, err)
 	assert.NotPanics(t, func() {
 		err := c.Write([]byte("abc"), b)
diff --git a/internal/storage/writer/s3/s3_test.go b/internal/storage/writer/s3/s3_test.go
index 65c04662..e31611a1 100644
--- a/internal/storage/writer/s3/s3_test.go
+++ b/internal/storage/writer/s3/s3_test.go
@@ -4,10 +4,14 @@
 package s3
 
 import (
+	"bytes"
 	"testing"
 
+	eorc "github.com/crphang/orc"
 	"github.com/kelindar/talaria/internal/encoding/block"
 	"github.com/kelindar/talaria/internal/encoding/key"
+	"github.com/kelindar/talaria/internal/encoding/orc"
+	"github.com/kelindar/talaria/internal/encoding/typeof"
 	"github.com/kelindar/talaria/internal/monitor"
 	"github.com/kelindar/talaria/internal/monitor/logging"
 	"github.com/kelindar/talaria/internal/monitor/statsd"
@@ -15,6 +19,30 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
+// Create a base block for testing purpose
+func blockBase() ([]block.Block, error) {
+	schema := typeof.Schema{
+		"col0": typeof.String,
+		"col1": typeof.Int64,
+		"col2": typeof.Float64,
+	}
+	orcSchema, _ := orc.SchemaFor(schema)
+
+	orcBuffer1 := &bytes.Buffer{}
+	writer, _ := eorc.NewWriter(orcBuffer1,
+		eorc.SetSchema(orcSchema))
+	_ = writer.Write("eventName", 1, 1.0)
+	_ = writer.Close()
+
+	apply := block.Transform(nil)
+
+	block, err := block.FromOrcBy(orcBuffer1.Bytes(), "col0", nil, apply)
+	if err != nil {
+		return nil, err
+	}
+	return block, nil
+}
+
 func TestS3Writer(t *testing.T) {
 	m := monitor.New(logging.NewNoop(), statsd.NewNoop(), "x", "y")
 	_, err := New(m, "testBucket", "", "us-east-1", "", "", "", "", "", "", 128)
@@ -32,7 +60,7 @@ func TestS3Writer_Write(t *testing.T) {
 	s3Writer.uploader = mockUploader
 	s3Writer.bucket = "testBucket"
 
-	block, err := block.Base()
+	block, err := blockBase()
 	assert.Nil(t, err)
 	err = s3Writer.Write(key.Key("testKey"), block)