feature:ParquetSource operator is implemented

Author: Richard Baah

.gitignore

@@ -106,4 +106,7 @@ src/Backend/test_data/json
 
 # Allow a specific CSV dataset that we want tracked despite the general csv ignores
 !src/Backend/test_data/csv/
-!src/Backend/test_data/csv/Mental_Health_and_Social_Media_Balance_Dataset.csv
+!src/Backend/test_data/csv/Mental_Health_and_Social_Media_Balance_Dataset.csv
+# allow parquet file
+!src/Backend/test_data/parquet/
+!src/Backend/test_data/parquet/capitals_clean.parquet

src/Backend/opti-sql-go/config/config.go

@@ -32,8 +32,9 @@ type batchConfig struct {
 	EnableParallelRead   bool   `yaml:"enable_parallel_read"`
 	MaxMemoryBeforeSpill uint64 `yaml:"max_memory_before_spill"`
 	MaxFileSizeMB        int    `yaml:"max_file_size_mb"` // max size of a single file
-	ShouldDowndload      bool   `yaml:"should_download"`
-	MaxDownloadSizeMB    int    `yaml:"max_download_size_mb"` // max size to download from external sources like S3
+	// TODO: add test for these two fileds, just add to existing test
+	ShouldDowndload   bool `yaml:"should_download"`
+	MaxDownloadSizeMB int  `yaml:"max_download_size_mb"` // max size to download from external sources like S3
 }
 type queryConfig struct {
 	// should results be cached, server side? if so how long

src/Backend/opti-sql-go/go.mod

@@ -11,6 +11,7 @@ require (
 )
 
 require (
+	github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c // indirect
 	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/apache/thrift v0.20.0 // indirect
 	github.com/goccy/go-json v0.10.3 // indirect
@@ -21,6 +22,7 @@ require (
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
 	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	golang.org/x/exp v0.0.0-20240222234643-814bf88cf225 // indirect
 	golang.org/x/mod v0.18.0 // indirect

src/Backend/opti-sql-go/go.sum

@@ -1,3 +1,5 @@
+github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c h1:RGWPOewvKIROun94nF7v2cua9qP+thov/7M50KEoeSU=
+github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c/go.mod h1:X0CRv0ky0k6m906ixxpzmDRLvX58TFUKS2eePweuyxk=
 github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
 github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
 github.com/apache/arrow/go/v15 v15.0.2 h1:60IliRbiyTWCWjERBCkO1W4Qun9svcYoZrSLcyOsMLE=

src/Backend/opti-sql-go/operators/filter/filter.go

@@ -1,7 +1,24 @@
 package filter
 
-// handle Bitwise operations here as well
+import (
+	"github.com/apache/arrow/go/v17/arrow"
+	"github.com/apache/arrow/go/v17/arrow/array"
+)
 
-// OR
-// AND
-// NOT
+// FilterExpr takes in a field and column and yeildss a function that takes in an index and returns a bool indicating whether the row at that index satisfies the filter condition.
+type FilterExpr func(filed arrow.Field, col arrow.Array) func(i int) bool
+
+// example
+func ExampleFilterExpr(field arrow.Field, col arrow.Array) func(i int) bool {
+	{
+		if field.Name == "age" && col.DataType().ID() == arrow.INT32 {
+			return func(i int) bool {
+				val := col.(*array.Int32).Value(i)
+				return val > 30
+			}
+		}
+		return func(i int) bool {
+			return true
+		}
+	}
+}

src/Backend/opti-sql-go/operators/project/source/csv.go

@@ -13,7 +13,9 @@ import (
 	"github.com/apache/arrow/go/v17/arrow/array"
 )
 
-type ProjectCSVLeaf struct {
+// TODO: change the leaf stuff to be called scans instead
+
+type CSVSource struct {
 	r            *csv.Reader
 	schema       *arrow.Schema // columns to project as well as types to cast to
 	colPosition  map[string]int
@@ -22,9 +24,9 @@ type ProjectCSVLeaf struct {
 }
 
 // assume everything is on disk for now
-func NewProjectCSVLeaf(source io.Reader) (*ProjectCSVLeaf, error) {
+func NewProjectCSVLeaf(source io.Reader) (*CSVSource, error) {
 	r := csv.NewReader(source)
-	proj := &ProjectCSVLeaf{
+	proj := &CSVSource{
 		r:           r,
 		colPosition: make(map[string]int),
 	}
@@ -34,31 +36,32 @@ func NewProjectCSVLeaf(source io.Reader) (*ProjectCSVLeaf, error) {
 	return proj, err
 }
 
-func (pcsv *ProjectCSVLeaf) Next(n uint64) (*operators.RecordBatch, error) {
-	if pcsv.done {
+func (csvS *CSVSource) Next(n uint64) (*operators.RecordBatch, error) {
+	if csvS.done {
 		return nil, io.EOF
 	}
 
 	// 1. Create builders
-	builders := pcsv.initBuilders()
+	builders := csvS.initBuilders()
 
 	rowsRead := uint64(0)
 
 	// Process stored first row (from parseHeader) ---
-	if pcsv.firstDataRow != nil && rowsRead < n {
-		if err := pcsv.processRow(pcsv.firstDataRow, builders); err != nil {
+	if csvS.firstDataRow != nil && rowsRead < n {
+		fmt.Printf("First row: %v\n", csvS.firstDataRow)
+		if err := csvS.processRow(csvS.firstDataRow, builders); err != nil {
 			return nil, err
 		}
-		pcsv.firstDataRow = nil // consume it once
+		csvS.firstDataRow = nil // consume it once
 		rowsRead++
 	}
 
 	//  Stream remaining rows from CSV reader ---
 	for rowsRead < n {
-		row, err := pcsv.r.Read()
+		row, err := csvS.r.Read()
 		if err == io.EOF {
 			if rowsRead == 0 {
-				pcsv.done = true
+				csvS.done = true
 				return nil, io.EOF
 			}
 			break
@@ -68,24 +71,24 @@ func (pcsv *ProjectCSVLeaf) Next(n uint64) (*operators.RecordBatch, error) {
 		}
 
 		// append to builders
-		if err := pcsv.processRow(row, builders); err != nil {
+		if err := csvS.processRow(row, builders); err != nil {
 			return nil, err
 		}
 
 		rowsRead++
 	}
 
 	//  Freeze into Arrow arrays
-	columns := pcsv.finalizeBuilders(builders)
+	columns := csvS.finalizeBuilders(builders)
 
 	return &operators.RecordBatch{
-		Schema:  pcsv.schema,
+		Schema:  csvS.schema,
 		Columns: columns,
 	}, nil
 }
 
-func (pcsv *ProjectCSVLeaf) initBuilders() []array.Builder {
-	fields := pcsv.schema.Fields()
+func (csvS *CSVSource) initBuilders() []array.Builder {
+	fields := csvS.schema.Fields()
 	builders := make([]array.Builder, len(fields))
 
 	for i, f := range fields {
@@ -94,14 +97,14 @@ func (pcsv *ProjectCSVLeaf) initBuilders() []array.Builder {
 
 	return builders
 }
-func (pcsv *ProjectCSVLeaf) processRow(
+func (csvS *CSVSource) processRow(
 	content []string,
 	builders []array.Builder,
 ) error {
-	fields := pcsv.schema.Fields()
-
+	fields := csvS.schema.Fields()
+	fmt.Printf("content : %v\n", content)
 	for i, f := range fields {
-		colIdx := pcsv.colPosition[f.Name]
+		colIdx := csvS.colPosition[f.Name]
 		cell := content[colIdx]
 
 		switch b := builders[i].(type) {
@@ -143,7 +146,7 @@ func (pcsv *ProjectCSVLeaf) processRow(
 
 	return nil
 }
-func (pcsv *ProjectCSVLeaf) finalizeBuilders(builders []array.Builder) []arrow.Array {
+func (csvS *CSVSource) finalizeBuilders(builders []array.Builder) []arrow.Array {
 	columns := make([]arrow.Array, len(builders))
 
 	for i, b := range builders {
@@ -155,16 +158,16 @@ func (pcsv *ProjectCSVLeaf) finalizeBuilders(builders []array.Builder) []arrow.A
 }
 
 // first call to csv.Reader
-func (pscv *ProjectCSVLeaf) parseHeader() (*arrow.Schema, error) {
-	header, err := pscv.r.Read()
+func (csvS *CSVSource) parseHeader() (*arrow.Schema, error) {
+	header, err := csvS.r.Read()
 	if err != nil {
 		return nil, err
 	}
-	firstDataRow, err := pscv.r.Read()
+	firstDataRow, err := csvS.r.Read()
 	if err != nil {
 		return nil, err
 	}
-	pscv.firstDataRow = firstDataRow
+	csvS.firstDataRow = firstDataRow
 	newFields := make([]arrow.Field, 0, len(header))
 	for i, colName := range header {
 		sampleValue := firstDataRow[i]
@@ -173,7 +176,7 @@ func (pscv *ProjectCSVLeaf) parseHeader() (*arrow.Schema, error) {
 			Type:     parseDataType(sampleValue),
 			Nullable: true,
 		})
-		pscv.colPosition[colName] = i
+		csvS.colPosition[colName] = i
 	}
 	return arrow.NewSchema(newFields, nil), nil
 }

src/Backend/opti-sql-go/operators/project/source/csv_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/apache/arrow/go/v17/arrow"
 	"github.com/apache/arrow/go/v17/arrow/array"
+	"github.com/apache/arrow/go/v17/arrow/memory"
 )
 
 const csvFilePath = "../../../../test_data/csv/Mental_Health_and_Social_Media_Balance_Dataset.csv"
@@ -943,6 +944,29 @@ func TestIntegrationWithRealFile(t *testing.T) {
 		}
 	})
 }
+func TestProccessFirstLine(t *testing.T) {
+	v := getTestFile()
+	p, err := NewProjectCSVLeaf(v)
+	if err != nil {
+		t.Errorf("Failed to create ProjectCSVLeaf: %v", err)
+	}
+	defer func() {
+		if err := v.Close(); err != nil {
+			t.Fatalf("failed to close: %v", err)
+		}
+	}()
+	var builders []array.Builder
+	for range len(p.schema.Fields()) {
+		builder := array.NewBuilder(memory.DefaultAllocator, &arrow.Date64Type{})
+		defer builder.Release()
+		builders = append(builders, builder)
+	}
+	err = p.processRow([]string{"1", "alice", "95.5", "true"}, builders)
+	if err == nil {
+		t.Errorf("Expected error for empty row, got nil")
+	}
+
+}
 
 /*
 func TestLargercsvFile(t *testing.T) {

src/Backend/opti-sql-go/operators/project/source/custom.go

@@ -22,14 +22,14 @@ var (
 	}
 )
 
-type InMemoryProjectExec struct {
+type InMemorySource struct {
 	schema        *arrow.Schema
 	columns       []arrow.Array
 	pos           uint64
 	fieldToColIDx map[string]int
 }
 
-func NewInMemoryProjectExec(names []string, columns []any) (*InMemoryProjectExec, error) {
+func NewInMemoryProjectExec(names []string, columns []any) (*InMemorySource, error) {
 	if len(names) != len(columns) {
 		return nil, operators.ErrInvalidSchema("number of column names and columns do not match")
 	}
@@ -49,51 +49,50 @@ func NewInMemoryProjectExec(names []string, columns []any) (*InMemoryProjectExec
 		arrays = append(arrays, arr)
 		fieldToColIDx[field.Name] = i
 	}
-	return &InMemoryProjectExec{
+	return &InMemorySource{
 		schema:        arrow.NewSchema(fields, nil),
 		columns:       arrays,
 		fieldToColIDx: fieldToColIDx,
 	}, nil
 }
-func (ime *InMemoryProjectExec) withFields(names ...string) error {
+func (ms *InMemorySource) withFields(names ...string) error {
 
-	newSchema, cols, err := project.ProjectSchemaFilterDown(ime.schema, ime.columns, names...)
+	newSchema, cols, err := project.ProjectSchemaFilterDown(ms.schema, ms.columns, names...)
 	if err != nil {
 		return err
 	}
 	newMap := make(map[string]int)
 	for i, f := range newSchema.Fields() {
 		newMap[f.Name] = i
-		fmt.Printf("%s:%d", f.Name, i)
 	}
-	ime.schema = newSchema
-	ime.fieldToColIDx = newMap
-	ime.columns = cols
+	ms.schema = newSchema
+	ms.fieldToColIDx = newMap
+	ms.columns = cols
 	return nil
 }
-func (ime *InMemoryProjectExec) Next(n uint64) (*operators.RecordBatch, error) {
-	if ime.pos >= uint64(ime.columns[0].Len()) {
+func (ms *InMemorySource) Next(n uint64) (*operators.RecordBatch, error) {
+	if ms.pos >= uint64(ms.columns[0].Len()) {
 		return nil, io.EOF // EOF
 	}
 	var currRows uint64 = 0
-	outPutCols := make([]arrow.Array, len(ime.schema.Fields()))
+	outPutCols := make([]arrow.Array, len(ms.schema.Fields()))
 
-	for i, field := range ime.schema.Fields() {
-		col := ime.columns[ime.fieldToColIDx[field.Name]]
+	for i, field := range ms.schema.Fields() {
+		col := ms.columns[ms.fieldToColIDx[field.Name]]
 		colLen := uint64(col.Len())
-		remaining := colLen - ime.pos
+		remaining := colLen - ms.pos
 		toRead := n
 		if remaining < n {
 			toRead = remaining
 		}
-		slice := array.NewSlice(col, int64(ime.pos), int64(ime.pos+toRead))
+		slice := array.NewSlice(col, int64(ms.pos), int64(ms.pos+toRead))
 		outPutCols[i] = slice
 		currRows = toRead
 	}
-	ime.pos += currRows
+	ms.pos += currRows
 
 	return &operators.RecordBatch{
-		Schema:  ime.schema,
+		Schema:  ms.schema,
 		Columns: outPutCols,
 	}, nil
 }