Skip to content

Commit 153607b

Browse files
fuziontechclaude
andauthored
fix: log multi-statement batches and add data_inlining_row_limit config (#367)
Fix a logging gap: executeSingleStatement (used for multi-statement batches like sqlmesh's "BEGIN; ALTER TABLE; INSERT; COMMIT") had zero logQuery calls. All statements in batches are now logged with protocol="simple-batch", making sqlmesh operations visible in system.query_log. Also adds a configurable data_inlining_row_limit option for DuckLake, settable via YAML config or DUCKGRES_DUCKLAKE_DATA_INLINING_ROW_LIMIT env var. Set to 0 to disable data inlining (new default in DuckLake 0.4). Negative values are rejected with a warning. Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 91a0b2d commit 153607b

6 files changed

Lines changed: 79 additions & 4 deletions

File tree

‎config_resolution.go‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -271,6 +271,13 @@ func resolveEffectiveConfig(fileCfg *FileConfig, cli configCLIInputs, getenv fun
271271
warn("Invalid ducklake.checkpoint_interval duration: " + err.Error())
272272
}
273273
}
274+
if fileCfg.DuckLake.DataInliningRowLimit != nil {
275+
if *fileCfg.DuckLake.DataInliningRowLimit < 0 {
276+
warn("ducklake.data_inlining_row_limit must be >= 0")
277+
} else {
278+
cfg.DuckLake.DataInliningRowLimit = fileCfg.DuckLake.DataInliningRowLimit
279+
}
280+
}
274281

275282
cfg.ProcessIsolation = fileCfg.ProcessIsolation
276283
if fileCfg.IdleTimeout != "" {
@@ -503,6 +510,17 @@ func resolveEffectiveConfig(fileCfg *FileConfig, cli configCLIInputs, getenv fun
503510
warn("Invalid DUCKGRES_DUCKLAKE_CHECKPOINT_INTERVAL duration: " + err.Error())
504511
}
505512
}
513+
if v := getenv("DUCKGRES_DUCKLAKE_DATA_INLINING_ROW_LIMIT"); v != "" {
514+
if n, err := strconv.Atoi(v); err == nil {
515+
if n < 0 {
516+
warn("DUCKGRES_DUCKLAKE_DATA_INLINING_ROW_LIMIT must be >= 0")
517+
} else {
518+
cfg.DuckLake.DataInliningRowLimit = &n
519+
}
520+
} else {
521+
warn("Invalid DUCKGRES_DUCKLAKE_DATA_INLINING_ROW_LIMIT: " + err.Error())
522+
}
523+
}
506524
if v := getenv("DUCKGRES_PROCESS_ISOLATION"); v != "" {
507525
if b, err := strconv.ParseBool(v); err == nil {
508526
cfg.ProcessIsolation = b

‎main.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,10 @@ type DuckLakeFileConfig struct {
125125

126126
// Checkpoint interval for DuckLake maintenance (e.g., "24h", "6h")
127127
CheckpointInterval string `yaml:"checkpoint_interval"`
128+
129+
// DataInliningRowLimit controls max rows inlined in metadata per insert.
130+
// Set to 0 to disable inlining. Omit to use DuckLake default (10).
131+
DataInliningRowLimit *int `yaml:"data_inlining_row_limit"`
128132
}
129133

130134
// loadConfigFile loads configuration from a YAML file

‎server/conn.go‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1252,6 +1252,8 @@ func (c *clientConn) handleMultiStatementQuery(tree *pg_query.ParseResult) error
12521252
// client (so the caller can stop processing a batch), or (false, err) for
12531253
// fatal connection errors.
12541254
func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalErr error) {
1255+
start := time.Now()
1256+
12551257
// Check for cursor operations before transpilation
12561258
tree, parseErr := pg_query.Parse(query)
12571259
if parseErr == nil && len(tree.Stmts) == 1 {
@@ -1466,20 +1468,29 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
14661468
}
14671469
}
14681470
if err != nil {
1471+
errCode := "42000"
1472+
errMsg := err.Error()
14691473
if isQueryCancelled(err) {
1470-
c.sendError("ERROR", "57014", "canceling statement due to user request")
1474+
errCode = "57014"
1475+
errMsg = "canceling statement due to user request"
14711476
} else {
14721477
slog.Error("Query execution failed.", "user", c.username, "query", executedQuery, "error", err)
1473-
c.sendError("ERROR", "42000", err.Error())
14741478
}
1479+
c.sendError("ERROR", errCode, errMsg)
14751480
c.setTxError()
1481+
c.logQuery(start, query, executedQuery, cmdType, 0, 0, errCode, errMsg, "simple-batch")
14761482
return true, nil
14771483
}
14781484
}
14791485

1486+
var writtenRows int64
1487+
if execResult != nil {
1488+
writtenRows, _ = execResult.RowsAffected()
1489+
}
14801490
c.updateTxStatus(cmdType)
14811491
tag := c.buildCommandTag(cmdType, execResult)
14821492
_ = writeCommandComplete(c.writer, tag)
1493+
c.logQuery(start, query, executedQuery, cmdType, 0, writtenRows, "", "", "simple-batch")
14831494
return false, nil
14841495
}
14851496

@@ -1489,13 +1500,17 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
14891500

14901501
rows, err := c.executor.QueryContext(ctx, executedQuery)
14911502
if err != nil {
1503+
errCode := "42000"
1504+
errMsg := err.Error()
14921505
if isQueryCancelled(err) {
1493-
c.sendError("ERROR", "57014", "canceling statement due to user request")
1506+
errCode = "57014"
1507+
errMsg = "canceling statement due to user request"
14941508
} else {
14951509
slog.Error("Query execution failed.", "user", c.username, "query", executedQuery, "error", err)
1496-
c.sendError("ERROR", "42000", err.Error())
14971510
}
1511+
c.sendError("ERROR", errCode, errMsg)
14981512
c.setTxError()
1513+
c.logQuery(start, query, executedQuery, cmdType, 0, 0, errCode, errMsg, "simple-batch")
14991514
return true, nil
15001515
}
15011516
defer func() { _ = rows.Close() }()
@@ -1546,6 +1561,7 @@ func (c *clientConn) executeSingleStatement(query string) (errSent bool, fatalEr
15461561
c.updateTxStatus(cmdType)
15471562
tag := buildCommandTagFromRowCount(cmdType, int64(rowCount))
15481563
_ = writeCommandComplete(c.writer, tag)
1564+
c.logQuery(start, query, executedQuery, cmdType, int64(rowCount), 0, "", "", "simple-batch")
15491565
return false, nil
15501566
}
15511567

‎server/ducklake_migration.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -531,6 +531,9 @@ func buildDuckLakeAttachStmt(dlCfg DuckLakeConfig, migrate bool) string {
531531
if migrate {
532532
options = append(options, "AUTOMATIC_MIGRATION TRUE")
533533
}
534+
if dlCfg.DataInliningRowLimit != nil {
535+
options = append(options, fmt.Sprintf("DATA_INLINING_ROW_LIMIT %d", *dlCfg.DataInliningRowLimit))
536+
}
534537

535538
if len(options) > 0 {
536539
return fmt.Sprintf("ATTACH 'ducklake:%s' AS ducklake (%s)",

‎server/ducklake_migration_test.go‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,33 @@ func TestBuildDuckLakeAttachStmt(t *testing.T) {
7979
migrate: false,
8080
want: "ATTACH 'ducklake:postgres:host=localhost password=it''s_secret' AS ducklake",
8181
},
82+
{
83+
name: "with data inlining disabled",
84+
dlCfg: DuckLakeConfig{
85+
MetadataStore: "postgres:host=localhost dbname=dl",
86+
DataInliningRowLimit: intPtr(0),
87+
},
88+
migrate: false,
89+
want: "ATTACH 'ducklake:postgres:host=localhost dbname=dl' AS ducklake (DATA_INLINING_ROW_LIMIT 0)",
90+
},
91+
{
92+
name: "with data inlining custom limit",
93+
dlCfg: DuckLakeConfig{
94+
MetadataStore: "postgres:host=localhost dbname=dl",
95+
ObjectStore: "s3://bucket/path",
96+
DataInliningRowLimit: intPtr(100),
97+
},
98+
migrate: false,
99+
want: "ATTACH 'ducklake:postgres:host=localhost dbname=dl' AS ducklake (DATA_PATH 's3://bucket/path', DATA_INLINING_ROW_LIMIT 100)",
100+
},
101+
{
102+
name: "nil data inlining uses default",
103+
dlCfg: DuckLakeConfig{
104+
MetadataStore: "postgres:host=localhost dbname=dl",
105+
},
106+
migrate: false,
107+
want: "ATTACH 'ducklake:postgres:host=localhost dbname=dl' AS ducklake",
108+
},
82109
}
83110

84111
for _, tt := range tests {
@@ -91,6 +118,8 @@ func TestBuildDuckLakeAttachStmt(t *testing.T) {
91118
}
92119
}
93120

121+
func intPtr(n int) *int { return &n }
122+
94123
func TestFormatSQLValue(t *testing.T) {
95124
tests := []struct {
96125
name string

‎server/server.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -240,6 +240,11 @@ type DuckLakeConfig struct {
240240
// Set to 0 to disable. Default: 24h.
241241
CheckpointInterval time.Duration
242242

243+
// DataInliningRowLimit controls the maximum number of rows to inline
244+
// in DuckLake metadata instead of writing to Parquet files.
245+
// Set to pointer-to-0 to disable inlining. nil uses DuckLake default (10).
246+
DataInliningRowLimit *int
247+
243248
// Migrate is set by the control plane after running the migration check.
244249
// When true, AttachDuckLake uses AUTOMATIC_MIGRATION TRUE without
245250
// re-running the version check. This avoids redundant backups and

0 commit comments

Comments
 (0)