🐛 Fix issue parquet quasi-binary import #52

Before, when a column had the BYTE_ARRAY type, anyquery would still try to interpret it at UTF-8, leading to a Replacement Character.

To fix this, Anyquery now make sure a []byte slice is returned rather than a string.

I couldn't find any other way to fix this as parquet-go/parquet-go returns a string even for a parquet BYTE_ARRAY

Author: julien040

go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/mark3labs/mcp-go v0.31.0
 	github.com/mattn/go-sqlite3 v1.14.28
 	github.com/olekukonko/tablewriter v0.0.5
-	github.com/parquet-go/parquet-go v0.25.0
+	github.com/parquet-go/parquet-go v0.25.1
 	github.com/runreveal/pql v0.2.0
 	github.com/samber/lo v1.50.0
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1

go.sum

@@ -1231,8 +1231,8 @@ github.com/opentracing/opentracing-go v1.2.0 h1:uEJPy/1a5RIPAJ0Ov+OIO8OxWu77jEv+
 github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYrxe9dPLANfrWvHYVTgc=
 github.com/outcaste-io/ristretto v0.2.3 h1:AK4zt/fJ76kjlYObOeNwh4T3asEuaCmp26pOvUOL9w0=
 github.com/outcaste-io/ristretto v0.2.3/go.mod h1:W8HywhmtlopSB1jeMg3JtdIhf+DYkLAr0VN/s4+MHac=
-github.com/parquet-go/parquet-go v0.25.0 h1:GwKy11MuF+al/lV6nUsFw8w8HCiPOSAx1/y8yFxjH5c=
-github.com/parquet-go/parquet-go v0.25.0/go.mod h1:OqBBRGBl7+llplCvDMql8dEKaDqjaFA/VAPw+OJiNiw=
+github.com/parquet-go/parquet-go v0.25.1 h1:l7jJwNM0xrk0cnIIptWMtnSnuxRkwq53S+Po3KG8Xgo=
+github.com/parquet-go/parquet-go v0.25.1/go.mod h1:AXBuotO1XiBtcqJb/FKFyjBG4aqa3aQAAWF3ZPzCanY=
 github.com/paulmach/orb v0.11.1 h1:3koVegMC4X/WeiXYz9iswopaTwMem53NzTJuTF20JzU=
 github.com/paulmach/orb v0.11.1/go.mod h1:5mULz1xQfs3bmQm63QEJA6lNGujuRafwA5S/EnuLaLU=
 github.com/paulmach/protoscan v0.2.1/go.mod h1:SpcSwydNLrxUGSDvXvO0P7g7AuhJ7lcKfDlhJCDw2gY=

module/read_parquet.go

@@ -18,11 +18,11 @@ type ParquetModule struct {
 
 type ParquetTable struct {
 	mmap   mmap.MMap
-	column map[int]string
+	column map[int]parquetColumn
 }
 
 type ParquetCursor struct {
-	column    map[int]string
+	column    map[int]parquetColumn
 	reader    *parquet.GenericReader[any]
 	rowBuffer *deque.Deque[map[string]interface{}]
 	rowID     int64
@@ -34,6 +34,12 @@ type ParquetCursor struct {
 	noMoreRows bool
 }
 
+type parquetColumn struct {
+	Name      string
+	Type      string
+	SubFields map[string]parquetColumn
+}
+
 const rowToRequestPerBatch = 16
 
 func (m *ParquetModule) Create(c *sqlite3.SQLiteConn, args []string) (sqlite3.VTab, error) {
@@ -69,7 +75,7 @@ func (m *ParquetModule) Connect(c *sqlite3.SQLiteConn, args []string) (sqlite3.V
 	}
 
 	// Open the file
-	mmap := mmap.MMap{}
+	var mmap mmap.MMap
 	var err error
 
 	mmap, err = openMmapedFile(fileName)
@@ -82,7 +88,7 @@ func (m *ParquetModule) Connect(c *sqlite3.SQLiteConn, args []string) (sqlite3.V
 	// Read the parquet file
 	reader := parquet.NewGenericReader[any](byteReader)
 
-	column := make(map[int]string)
+	column := make(map[int]parquetColumn)
 
 	sqlSchema := strings.Builder{}
 	sqlSchema.WriteString("CREATE TABLE parquet (")
@@ -106,9 +112,28 @@ func (m *ParquetModule) Connect(c *sqlite3.SQLiteConn, args []string) (sqlite3.V
 		default:
 			sqlSchema.WriteString("TEXT")
 		}
+
 		// Save the column name
-		column[i] = field.Name()
+		col := parquetColumn{
+			Name: field.Name(),
+			Type: field.Type().String(),
+		}
+
+		// Get subfields if the field is a group
+		if field.Type().String() == "group" {
+			col.SubFields = make(map[string]parquetColumn)
+			for _, subField := range field.Fields() {
+				col.SubFields[subField.Name()] = parquetColumn{
+					Name: subField.Name(),
+					Type: subField.Type().String(),
+				}
+			}
+		}
+
+		// Save the column in the map
+		column[i] = col
 	}
+
 	sqlSchema.WriteString(");")
 	c.DeclareVTab(sqlSchema.String())
 
@@ -118,6 +143,7 @@ func (m *ParquetModule) Connect(c *sqlite3.SQLiteConn, args []string) (sqlite3.V
 func (t *ParquetTable) Open() (sqlite3.VTabCursor, error) {
 	// Create a new reader
 	reader := parquet.NewGenericReader[any](bytes.NewReader(t.mmap))
+
 	return &ParquetCursor{
 		column:    t.column,
 		reader:    reader,
@@ -195,7 +221,7 @@ func (t *ParquetCursor) Column(context *sqlite3.SQLiteContext, col int) error {
 		context.ResultNull()
 		return nil
 	}
-	val, ok := t.rowBuffer.Front()[colName]
+	val, ok := t.rowBuffer.Front()[colName.Name]
 	if !ok {
 		context.ResultNull()
 		return nil
@@ -225,10 +251,36 @@ func (t *ParquetCursor) Column(context *sqlite3.SQLiteContext, col int) error {
 	case float64:
 		context.ResultDouble(valParsed)
 	case string:
-		context.ResultText(valParsed)
+		// parquet-go returns BYTE_ARRAY as string resulting in UTF-8 issues
+		// When we detect a column that has a BYTE_ARRAY type, we will convert it to a byte slice
+		if colName.Type == "BYTE_ARRAY" || colName.Type == "FIXED_LEN_BYTE_ARRAY" {
+			// Convert the string to a byte slice
+			context.ResultBlob([]byte(valParsed))
+		} else {
+			context.ResultText(valParsed)
+		}
 	case []byte:
 		context.ResultBlob(valParsed)
 	case map[string]interface{}:
+		for key, value := range valParsed {
+
+			// Get the subfield type if it exists
+			subFieldType, ok := colName.SubFields[key]
+			if !ok {
+				continue
+			}
+
+			// Same as the string case, we need to handle BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY
+			if subFieldType.Type == "BYTE_ARRAY" || subFieldType.Type == "FIXED_LEN_BYTE_ARRAY" {
+				if strValue, ok := value.(string); ok {
+					valParsed[key] = []byte(strValue)
+				} else if byteValue, ok := value.([]byte); ok {
+					valParsed[key] = byteValue
+				} else {
+					valParsed[key] = value
+				}
+			}
+		}
 		marshaled, err := json.Marshal(valParsed)
 		if err != nil {
 			context.ResultNull()