cache schema fields to avoid memory allocation on every Next() call

Author: s0und0fs1lence

chdb/driver/driver.go

@@ -54,6 +54,7 @@ func (d DriverType) PrepareRows(result chdbpurego.ChdbResult, buf []byte, bufSiz
 			localResult: result, reader: reader,
 			bufferSize: bufSize, needNewBuffer: true,
 			useUnsafeStringReader: useUnsafe,
+			schemaFields:          reader.Schema().Fields(),
 		}, nil
 
 	}
@@ -73,6 +74,7 @@ func (d DriverType) PrepareStreamingRows(result chdbpurego.ChdbStreamResult, buf
 			stream: result, curChunk: nextRes, reader: reader,
 			bufferSize: bufSize, needNewBuffer: true,
 			useUnsafeStringReader: useUnsafe,
+			schemaFields:          reader.Schema().Fields(),
 		}, nil
 
 	}

chdb/driver/parquet.go

@@ -28,6 +28,7 @@ type parquetRows struct {
 	reader                *parquet.GenericReader[any] // parquet reader
 	curRecord             parquet.Row                 // TODO: delete this?
 	buffer                []parquet.Row               // record buffer
+	schemaFields          []parquet.Field             // schema fields
 	bufferSize            int                         // amount of records to preload into buffer
 	bufferIndex           int64                       // index in the current buffer
 	curRow                int64                       // row counter
@@ -36,8 +37,7 @@ type parquetRows struct {
 }
 
 func (r *parquetRows) Columns() (out []string) {
-	sch := r.reader.Schema()
-	for _, f := range sch.Fields() {
+	for _, f := range r.schemaFields {
 		out = append(out, f.Name())
 	}
 
@@ -53,7 +53,7 @@ func (r *parquetRows) Close() error {
 	r.reader = nil
 	r.localResult.Free()
 	r.localResult = nil
-
+	r.schemaFields = nil
 	r.buffer = nil
 	return nil
 }
@@ -90,7 +90,7 @@ func (r *parquetRows) Next(dest []driver.Value) error {
 
 	}
 	r.curRecord = r.buffer[r.bufferIndex]
-	if r.curRecord == nil || len(r.curRecord) == 0 {
+	if len(r.curRecord) == 0 {
 		return fmt.Errorf("empty row")
 	}
 	var scanError error
@@ -166,19 +166,19 @@ func (r *parquetRows) Next(dest []driver.Value) error {
 }
 
 func (r *parquetRows) ColumnTypeDatabaseTypeName(index int) string {
-	return r.reader.Schema().Fields()[index].Type().String()
+	return r.schemaFields[index].Type().String()
 }
 
 func (r *parquetRows) ColumnTypeNullable(index int) (nullable, ok bool) {
-	return r.reader.Schema().Fields()[index].Optional(), true
+	return r.schemaFields[index].Optional(), true
 }
 
 func (r *parquetRows) ColumnTypePrecisionScale(index int) (precision, scale int64, ok bool) {
 	return 0, 0, false
 }
 
 func (r *parquetRows) ColumnTypeScanType(index int) reflect.Type {
-	switch r.reader.Schema().Fields()[index].Type().Kind() {
+	switch r.schemaFields[index].Type().Kind() {
 	case parquet.Boolean:
 		return reflect.TypeOf(false)
 	case parquet.Int32:

chdb/driver/parquet_streaming.go

@@ -18,17 +18,17 @@ type parquetStreamingRows struct {
 	curChunk              chdbpurego.ChdbResult       // current chunk
 	reader                *parquet.GenericReader[any] // parquet reader
 	curRecord             parquet.Row
-	buffer                []parquet.Row // record buffer
-	bufferSize            int           // amount of records to preload into buffer
-	bufferIndex           int64         // index in the current buffer
-	curRow                int64         // row counter
+	buffer                []parquet.Row   // record buffer
+	schemaFields          []parquet.Field // schema fields
+	bufferSize            int             // amount of records to preload into buffer
+	bufferIndex           int64           // index in the current buffer
+	curRow                int64           // row counter
 	needNewBuffer         bool
 	useUnsafeStringReader bool
 }
 
 func (r *parquetStreamingRows) Columns() (out []string) {
-	sch := r.reader.Schema()
-	for _, f := range sch.Fields() {
+	for _, f := range r.schemaFields {
 		out = append(out, f.Name())
 	}
 
@@ -45,6 +45,7 @@ func (r *parquetStreamingRows) Close() error {
 	r.stream.Free()
 	r.curChunk = nil
 	r.stream = nil
+	r.schemaFields = nil
 
 	r.buffer = nil
 	return nil
@@ -85,6 +86,7 @@ func (r *parquetStreamingRows) readNextChunkFromStream() error {
 		return io.EOF
 	}
 	r.reader = parquet.NewGenericReader[any](bytes.NewReader(r.curChunk.Buf()))
+	r.schemaFields = r.reader.Schema().Fields()
 	return nil
 }
 
@@ -182,19 +184,19 @@ func (r *parquetStreamingRows) Next(dest []driver.Value) error {
 }
 
 func (r *parquetStreamingRows) ColumnTypeDatabaseTypeName(index int) string {
-	return r.reader.Schema().Fields()[index].Type().String()
+	return r.schemaFields[index].Type().String()
 }
 
 func (r *parquetStreamingRows) ColumnTypeNullable(index int) (nullable, ok bool) {
-	return r.reader.Schema().Fields()[index].Optional(), true
+	return r.schemaFields[index].Optional(), true
 }
 
 func (r *parquetStreamingRows) ColumnTypePrecisionScale(index int) (precision, scale int64, ok bool) {
 	return 0, 0, false
 }
 
 func (r *parquetStreamingRows) ColumnTypeScanType(index int) reflect.Type {
-	switch r.reader.Schema().Fields()[index].Type().Kind() {
+	switch r.schemaFields[index].Type().Kind() {
 	case parquet.Boolean:
 		return reflect.TypeOf(false)
 	case parquet.Int32: