Flush writes the accumulated changes to the database.
(ctx *sql.Context, conn *stdsql.Conn, tx *stdsql.Tx, reason FlushReason)
| 73 | |
| 74 | // Flush writes the accumulated changes to the database. |
| 75 | func (c *DeltaController) Flush(ctx *sql.Context, conn *stdsql.Conn, tx *stdsql.Tx, reason FlushReason) (FlushStats, error) { |
| 76 | c.mutex.Lock() |
| 77 | defer c.mutex.Unlock() |
| 78 | |
| 79 | // Due to DuckDB's lack of support for atomic MERGE INTO, we have to do the following two steps separately: |
| 80 | // 1. Delete rows that are being updated. |
| 81 | // 2. Insert new rows. |
| 82 | // To guarantee the atomicity of the two steps, we have to wrap them in a transaction. |
| 83 | // Again, due to DuckDB's limitations of indexes, specifically over-eagerly unique constraint checking, |
| 84 | // if we do **DELETE then INSERT** in the same transaction, we would get |
| 85 | // a unique constraint violation error for INSERT, |
| 86 | // or data corruption for INSERT OR REPLACE|IGNORE INTO. |
| 87 | // |
| 88 | // This is a noteworthy pitfall and seems unlikely to be fixed in DuckDB in the near future, |
| 89 | // but we have to live with it. |
| 90 | // |
| 91 | // On the other hand, fortunately enough, **INSERT OR REPLACE then DELETE** in the same transaction works fine. |
| 92 | // |
| 93 | // The ultimate solution is to wait for DuckDB to improve its index handling. |
| 94 | // In the meantime, we could contribute a patch to DuckDB to support atomic MERGE INTO, |
| 95 | // which is another way to avoid the issue elegantly. |
| 96 | // |
| 97 | // See: |
| 98 | // https://duckdb.org/docs/sql/indexes.html#limitations-of-art-indexes |
| 99 | // https://github.com/duckdb/duckdb/issues/14133 |
| 100 | |
| 101 | var stats FlushStats |
| 102 | |
| 103 | for table, appender := range c.tables { |
| 104 | deltaRowCount := appender.RowCount() |
| 105 | if deltaRowCount > 0 { |
| 106 | if err := c.updateTable(ctx, conn, tx, table, appender, &stats); err != nil { |
| 107 | return stats, err |
| 108 | } |
| 109 | } |
| 110 | switch reason { |
| 111 | case DDLStmtFlushReason: |
| 112 | // DDL statement may change the schema |
| 113 | delete(c.tables, table) |
| 114 | default: |
| 115 | // Pre-allocate memory for the next delta |
| 116 | if deltaRowCount > 0 { |
| 117 | // Next power of 2 |
| 118 | appender.Grow(1 << bits.Len64(uint64(deltaRowCount)-1)) |
| 119 | } |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | if stats.DeltaSize > 0 { |
| 124 | if log := ctx.GetLogger(); log.Logger.IsLevelEnabled(logrus.DebugLevel) { |
| 125 | log.WithFields(logrus.Fields{ |
| 126 | "DeltaSize": stats.DeltaSize, |
| 127 | "Insertions": stats.Insertions, |
| 128 | "Deletions": stats.Deletions, |
| 129 | "Reason": reason.String(), |
| 130 | }).Debug("Flushed delta buffer") |
| 131 | } |
| 132 | } |
no test coverage detected