Method handleGeneralCase

delta/controller.go:442–526

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

}

func (c *DeltaController) handleGeneralCase(
	ctx *sql.Context,
	conn *stdsql.Conn,
	tx *stdsql.Tx,
	table tableIdentifier,
	appender *DeltaAppender,
	stats *FlushStats,
) error {
	if err := c.materializeCondensedDelta(ctx, conn, tx, table, appender, stats); err != nil {
		return err
	}
	defer tx.ExecContext(ctx, "DROP TABLE IF EXISTS temp.main.delta")

	qualifiedTableName := catalog.ConnectIdentifiersANSI(table.dbName, table.tableName)
	affected := int64(0)

	// Insert or replace new rows (action = INSERT) into the base table.
	insertSQL := "INSERT OR REPLACE INTO " +
		qualifiedTableName +
		" SELECT * EXCLUDE (" + AugmentedColumnList + ") FROM temp.main.delta WHERE action = " +
		strconv.Itoa(int(binlog.InsertRowEvent))
	result, err := tx.ExecContext(ctx, insertSQL)
	if err == nil {
		affected, err = result.RowsAffected()
	}
	if err != nil {
		return err
	}
	stats.Insertions += affected

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

	// Delete rows that have been deleted.
	// The plan for `IN` is optimized to a SEMI JOIN,
	// which is more efficient than ordinary INNER JOIN.
	// DuckDB does not support multiple columns in `IN` clauses,
	// so we need to handle this case separately using the `row()` function.
	inTuple := getPrimaryKeyStruct(appender.BaseSchema())
	deleteSQL := "DELETE FROM " + qualifiedTableName +
		" WHERE " + inTuple + " IN (SELECT " + inTuple +
		"FROM temp.main.delta WHERE action = " + strconv.Itoa(int(binlog.DeleteRowEvent)) + ")"
	result, err = tx.ExecContext(ctx, deleteSQL)
	if err == nil {
		affected, err = result.RowsAffected()
	}
	if err != nil {
		return err
	}
	stats.Deletions += affected

	// For debugging:
	//