Method prepareArrowView

delta/controller.go:193–264

Helper function to build the Arrow record and register the view

(
	ctx *sql.Context,
	conn *stdsql.Conn,
	table tableIdentifier,
	appender *DeltaAppender,
	fieldOffset int,
	fieldIndices []int,
)

// Helper function to build the Arrow record and register the view
func (c *DeltaController) prepareArrowView(
	ctx *sql.Context,
	conn *stdsql.Conn,
	table tableIdentifier,
	appender *DeltaAppender,
	fieldOffset int,
	fieldIndices []int,
) (viewName string, close func(), err error) {
	record := appender.Build()

	// fmt.Println("record:", record)

	var ar *duckdb.Arrow
	err = conn.Raw(func(driverConn any) error {
		var err error
		ar, err = duckdb.NewArrowFromConn(driverConn.(*duckdb.Conn))
		return err
	})
	if err != nil {
		record.Release()
		return "", nil, err
	}

	// Project the fields before registering the Arrow record into DuckDB.
	// Currently, this is necessary because RegisterView uses `arrow_scan` instead of `arrow_scan_dumb` under the hood.
	// The former allows projection & filter pushdown, but the implementation does not work as expected in some cases.
	schema := record.Schema()
	if fieldOffset > 0 {
		fields := schema.Fields()[fieldOffset:]
		schema = arrow.NewSchema(fields, nil)
		columns := record.Columns()[fieldOffset:]
		projected := array.NewRecord(schema, columns, record.NumRows())
		record.Release()
		record = projected
	} else if len(fieldIndices) > 0 {
		fields := make([]arrow.Field, len(fieldIndices))
		columns := make([]arrow.Array, len(fieldIndices))
		for i, idx := range fieldIndices {
			fields[i] = schema.Field(idx)
			columns[i] = record.Column(idx)
		}
		schema = arrow.NewSchema(fields, nil)
		projected := array.NewRecord(schema, columns, record.NumRows())
		record.Release()
		record = projected
	}

	reader, err := array.NewRecordReader(schema, []arrow.Record{record})
	if err != nil {
		record.Release()
		return "", nil, err
	}

	// Register the Arrow view
	hash := maphash.String(c.seed, table.dbName+"\x00"+table.tableName)
	viewName = "__sys_view_arrow_delta_" + strconv.FormatUint(hash, 16) + "__"

	release, err := ar.RegisterView(reader, viewName)