backend_jupiter/internal/backup/mysql_to_parquet.go

package backup

import (
	"database/sql"
	"os"
	"time"

	"github.com/apache/arrow/go/v14/arrow"
	"github.com/apache/arrow/go/v14/arrow/array"
	"github.com/apache/arrow/go/v14/arrow/memory"
	"github.com/apache/arrow/go/v14/parquet"
	"github.com/apache/arrow/go/v14/parquet/compress"
	"github.com/apache/arrow/go/v14/parquet/pqarrow"
)

func BackupDeliveryLogs(sqlDB *sql.DB) (string, error) {

	rows, err := sqlDB.Query(`
		SELECT
			logid, logdate, deliveryid, orderheaderid,
			tenantid, locationid, userid, partnerid,
			orderid, orderstatus, latitude, longitude,
			logstatus, created, updated, firstname, lastname
		FROM deliverylogs
		WHERE logdate >= NOW() - INTERVAL '3 months'
	`)
	if err != nil {
		return "", err
	}
	defer rows.Close()

	// Arrow schema (exact match with MySQL table)
	schema := arrow.NewSchema([]arrow.Field{
		{Name: "logid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "logdate", Type: arrow.FixedWidthTypes.Timestamp_ms, Nullable: true},
		{Name: "deliveryid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "orderheaderid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "tenantid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "locationid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "userid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "partnerid", Type: arrow.PrimitiveTypes.Int32},
		{Name: "orderid", Type: arrow.BinaryTypes.String, Nullable: true},
		{Name: "orderstatus", Type: arrow.BinaryTypes.String, Nullable: true},
		{Name: "latitude", Type: arrow.BinaryTypes.String, Nullable: true},
		{Name: "longitude", Type: arrow.BinaryTypes.String, Nullable: true},
		{Name: "logstatus", Type: arrow.PrimitiveTypes.Int32},
		{Name: "created", Type: arrow.FixedWidthTypes.Timestamp_ms},
		{Name: "updated", Type: arrow.FixedWidthTypes.Timestamp_ms},
		{Name: "firstname", Type: arrow.BinaryTypes.String, Nullable: true},
		{Name: "lastname", Type: arrow.BinaryTypes.String, Nullable: true},
	}, nil)

	pool := memory.NewGoAllocator()
	builder := array.NewRecordBuilder(pool, schema)
	defer builder.Release()

	for rows.Next() {

		var (
			logid, deliveryid, orderheaderid int32
			tenantid, locationid, userid     int32
			partnerid, logstatus             int32

			logdate, created, updated sql.NullTime
			orderid, orderstatus      sql.NullString
			latitude, longitude       sql.NullString
			firstname, lastname       sql.NullString
		)

		if err := rows.Scan(
			&logid, &logdate, &deliveryid, &orderheaderid,
			&tenantid, &locationid, &userid, &partnerid,
			&orderid, &orderstatus, &latitude, &longitude,
			&logstatus, &created, &updated, &firstname, &lastname,
		); err != nil {
			return "", err
		}

		builder.Field(0).(*array.Int32Builder).Append(logid)

		if logdate.Valid {
			builder.Field(1).(*array.TimestampBuilder).
				Append(arrow.Timestamp(logdate.Time.UnixMilli()))
		} else {
			builder.Field(1).(*array.TimestampBuilder).AppendNull()
		}

		builder.Field(2).(*array.Int32Builder).Append(deliveryid)
		builder.Field(3).(*array.Int32Builder).Append(orderheaderid)
		builder.Field(4).(*array.Int32Builder).Append(tenantid)
		builder.Field(5).(*array.Int32Builder).Append(locationid)
		builder.Field(6).(*array.Int32Builder).Append(userid)
		builder.Field(7).(*array.Int32Builder).Append(partnerid)

		appendString(builder.Field(8), orderid)
		appendString(builder.Field(9), orderstatus)
		appendString(builder.Field(10), latitude)
		appendString(builder.Field(11), longitude)

		builder.Field(12).(*array.Int32Builder).Append(logstatus)

		builder.Field(13).(*array.TimestampBuilder).
			Append(arrow.Timestamp(created.Time.UnixMilli()))
		builder.Field(14).(*array.TimestampBuilder).
			Append(arrow.Timestamp(updated.Time.UnixMilli()))

		appendString(builder.Field(15), firstname)
		appendString(builder.Field(16), lastname)
	}

	record := builder.NewRecord()
	defer record.Release()

	_ = os.MkdirAll("backups/parquet/deliverylogs", 0755)

	filePath := "backups/parquet/deliverylogs/deliverylogs_last_3_months_" +
		time.Now().Format("2006_01_02") + ".parquet"

	file, err := os.Create(filePath)
	if err != nil {
		return "", err
	}
	defer file.Close()

	// ✅ Correct Parquet + Arrow writer props (Arrow v14)
	parquetProps := parquet.NewWriterProperties(
		parquet.WithCompression(compress.Codecs.Snappy),
	)

	writer, err := pqarrow.NewFileWriter(
		schema,
		file,
		parquetProps,
		pqarrow.ArrowWriterProperties{}, // zero-value is valid
	)
	if err != nil {
		return "", err
	}
	defer writer.Close()

	if err := writer.Write(record); err != nil {
		return "", err
	}

	return filePath, nil
}

// helper for nullable strings
func appendString(b array.Builder, v sql.NullString) {
	sb := b.(*array.StringBuilder)
	if v.Valid {
		sb.Append(v.String)
	} else {
		sb.AppendNull()
	}
}