Method handleInsertOnly

delta/controller.go:266–315

(
	ctx *sql.Context,
	conn *stdsql.Conn,
	tx *stdsql.Tx,
	table tableIdentifier,
	appender *DeltaAppender,
	stats *FlushStats,
)

}

func (c *DeltaController) handleInsertOnly(
	ctx *sql.Context,
	conn *stdsql.Conn,
	tx *stdsql.Tx,
	table tableIdentifier,
	appender *DeltaAppender,
	stats *FlushStats,
) error {
	// Ignore the augmented fields
	viewName, release, err := c.prepareArrowView(ctx, conn, table, appender, appender.NumAugmentedFields(), nil)
	if err != nil {
		return err
	}
	defer release()

	// Perform direct INSERT without deduplication
	var b strings.Builder
	b.Grow(128)

	b.WriteString("INSERT INTO ")
	b.WriteString(catalog.ConnectIdentifiersANSI(table.dbName, table.tableName))
	b.WriteString(" SELECT ")
	buildColumnList(&b, appender.BaseSchema())
	b.WriteString(" FROM ")
	b.WriteString(viewName)

	sql := b.String()

	result, err := tx.ExecContext(ctx, sql)
	if err != nil {
		return err
	}

	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	stats.Insertions += affected
	stats.DeltaSize += affected

	if log := ctx.GetLogger(); log.Logger.IsLevelEnabled(logrus.DebugLevel) {
		log.WithFields(logrus.Fields{
			"db":    table.dbName,
			"table": table.tableName,
			"rows":  affected,
		}).Debug("Inserted")
	}

	return nil
}

func (c *DeltaController) handleDeleteOnly(
	ctx *sql.Context,