diff --git a/storage/duckdb/common/duckdb_config.h b/storage/duckdb/common/duckdb_config.h index 3ca06a0083015..e6a421bab5072 100644 --- a/storage/duckdb/common/duckdb_config.h +++ b/storage/duckdb/common/duckdb_config.h @@ -67,6 +67,7 @@ ulonglong get_thd_merge_join_threshold(THD *thd); my_bool get_thd_force_no_collation(THD *thd); ulong get_thd_explain_output(THD *thd); ulonglong get_thd_disabled_optimizers(THD *thd); +my_bool get_thd_cross_engine_ryow(THD *thd); /* Explain output types */ enum enum_explain_output diff --git a/storage/duckdb/docs/architecture.md b/storage/duckdb/docs/architecture.md index 534f1e9fcf4c8..895b4eddf832c 100644 --- a/storage/duckdb/docs/architecture.md +++ b/storage/duckdb/docs/architecture.md @@ -94,6 +94,67 @@ ha_duckdb_select_handler::init_scan() The fiber runs on the same OS thread as DuckDB. TLS (`current_thd`, `THR_KEY_mysys`) is swapped around fiber spawn/continue calls. +#### Pushdown modes: `duckdb_cross_engine_ryow` + +The session variable `duckdb_cross_engine_ryow` (default `OFF`) selects, per +query, how external (non-DuckDB) tables are read. The flag is captured once in +`init_scan()` (`register_cross_engine_ryow()`), and the replacement scan then +redirects to one of two table functions. + +| Aspect | `duckdb_cross_engine_ryow = OFF` (default) | `duckdb_cross_engine_ryow = ON` | +|---|---|---| +| Table function | `_mdb_scan` | `_mdb_scan_direct` | +| Scan mechanism | fiber runs a synthetic `SELECT … WHERE …` via `mysql_execute_command()` | direct `ha_rnd_init` / `ha_rnd_next` on the parent handler | +| Transaction | separate background-THD transaction | parent THD's transaction | +| Visibility | committed data only — the fiber runs in a *separate transaction* (background THD), so the parent's uncommitted writes are **not** visible (isolation follows the server default) | **Read-Your-Own-Writes**: the parent's uncommitted writes are visible | +| MariaDB optimizer for the external table | yes — range/index access, ICP (`idx_cond_push`), range-filter pushdown, index choice | no — plain full table scan | +| DuckDB `filter_pushdown` | `true` — DuckDB removes the predicates from the plan; the fiber applies them via the WHERE registered by `make_cond_for_table()` | `false` — DuckDB keeps the `Filter` operator above the scan and applies predicates itself | +| DuckDB `projection_pushdown` | `true` | `true` | + +Why `filter_pushdown` differs: `PhysicalTableScan` returns the table function's +chunk **without re-applying** `table_filters`. When `filter_pushdown = true` +the optimizer strips those predicates and the scan is solely responsible for +them. The direct `ha_rnd_next` path does not evaluate DuckDB `input.filters`, +so `_mdb_scan_direct` sets `filter_pushdown = false` to keep correctness via +DuckDB's own `Filter` operator. Projection pushdown stays enabled in both modes +(the direct path honors `column_ids` through the read set). + +Tradeoffs of `ON`: RYOW at the cost of losing the external engine's range/index +access and `idx_cond_push` for that table (it becomes a full scan). Note that +`cond_push` (engine condition pushdown) is unavailable for InnoDB/Aria/MyISAM/ +RocksDB regardless of mode — only ICP is, and only in the `OFF` (fiber) path. + +#### Future: index access / `idx_cond_push` in the RYOW path + +The RYOW path currently loses row-count reduction because it does a full +`ha_rnd_next` scan. This can be improved without giving up RYOW or correctness: + +1. `input.filters` (a DuckDB `TableFilterSet`) is already available in + `mdb_scan_init_global`. Inspect it for sargable predicates + (`CONSTANT_COMPARISON`, `IS NULL`, `IN`) on a key prefix of some index of + the external `TABLE`. +2. Replace `ha_rnd_init` with `ha_index_init(keyno)` + `ha_index_read_map()` + and an `end_range`, iterating with `ha_index_next()`. This recovers range + access — the larger win — and needs no server `Item` tree. +3. Optionally, for residual index-column predicates, synthesize an `Item` + expression from the `TableFilter`s over the table's `Field*` and call + `tbl->file->idx_cond_push(keyno, item)`, so the engine checks them at the + index level. + +Key constraint: keep `filter_pushdown = false` even after this work. In DuckDB's +model the function must apply pushed filters *entirely* (no re-check), so a +partial/best-effort translation would be unsafe. Treating `input.filters` purely +as a seek/ICP *hint* — while DuckDB's `Filter` stays the source of truth — keeps +correctness independent of translation quality. + +Caveats: `Item` synthesis needs a THD + memory root (use `tbl->in_use`); +constant/collation coercion between the DuckDB value and the MariaDB `Field`; +only AND-of-per-column predicates are sargable (OR is not); the parent handler +is mid-statement (opened and RDLCK-locked), so switch cleanly between rnd/index +scans and keep a single active scan; InnoDB refuses ICP on indexes with virtual +columns (`idx_cond_push` returns the condition), leaving a residual that must +fall back to DuckDB's `Filter`. + ### Path 2: DDL via Handler API ``` diff --git a/storage/duckdb/ha_duckdb.cc b/storage/duckdb/ha_duckdb.cc index e1b526b3e0fbe..564de785cbafc 100644 --- a/storage/duckdb/ha_duckdb.cc +++ b/storage/duckdb/ha_duckdb.cc @@ -1297,6 +1297,13 @@ static MYSQL_THDVAR_SET(disabled_optimizers, PLUGIN_VAR_RQCMDARG, "Disable specific DuckDB optimizer rules", NULL, NULL, 0, &myduck::disabled_optimizers_typelib); +static MYSQL_THDVAR_BOOL(cross_engine_ryow, PLUGIN_VAR_RQCMDARG, + "In cross-engine joins, read own uncommitted writes " + "from non-DuckDB tables via a direct handler scan in " + "the parent transaction (disables predicate pushdown " + "and index access path for those tables)", + NULL, NULL, FALSE); + /* ---- THDVAR accessor functions (used from duckdb_context.cc) ---- */ namespace myduck @@ -1319,6 +1326,11 @@ ulonglong get_thd_disabled_optimizers(THD *thd) return THDVAR(thd, disabled_optimizers); } +my_bool get_thd_cross_engine_ryow(THD *thd) +{ + return THDVAR(thd, cross_engine_ryow); +} + } // namespace myduck static struct st_mysql_sys_var *duckdb_system_variables[]= { @@ -1335,7 +1347,8 @@ static struct st_mysql_sys_var *duckdb_system_variables[]= { MYSQL_SYSVAR(log_options), /* Session proxy */ MYSQL_SYSVAR(merge_join_threshold), MYSQL_SYSVAR(force_no_collation), - MYSQL_SYSVAR(explain_output), MYSQL_SYSVAR(disabled_optimizers), NULL}; + MYSQL_SYSVAR(explain_output), MYSQL_SYSVAR(disabled_optimizers), + MYSQL_SYSVAR(cross_engine_ryow), NULL}; static struct st_mysql_show_var duckdb_status_variables[]= { {"Duckdb_rows_insert", (char *) &srv_duckdb_status.duckdb_rows_insert, diff --git a/storage/duckdb/ha_duckdb_pushdown.cc b/storage/duckdb/ha_duckdb_pushdown.cc index 49514d51bb812..fc979e723db3e 100644 --- a/storage/duckdb/ha_duckdb_pushdown.cc +++ b/storage/duckdb/ha_duckdb_pushdown.cc @@ -30,6 +30,7 @@ #include "duckdb_query.h" #include "duckdb_context.h" #include "cross_engine_scan.h" +#include "duckdb_config.h" #include "duckdb_log.h" extern handlerton *duckdb_hton; @@ -355,6 +356,8 @@ int ha_duckdb_select_handler::init_scan() */ if (has_cross_engine) { + myduck::register_cross_engine_ryow(myduck::get_thd_cross_engine_ryow(thd)); + auto register_tables_from_sel= [this](SELECT_LEX *sl) { for (TABLE_LIST *tbl= sl->get_table_list(); tbl; tbl= tbl->next_global) { diff --git a/storage/duckdb/mysql-test/duckdb/r/cross_engine_ryow.result b/storage/duckdb/mysql-test/duckdb/r/cross_engine_ryow.result new file mode 100644 index 0000000000000..f1d7fcf30d0ee --- /dev/null +++ b/storage/duckdb/mysql-test/duckdb/r/cross_engine_ryow.result @@ -0,0 +1,74 @@ +# +# Cross-engine RYOW: duckdb_cross_engine_ryow toggles visibility of +# uncommitted parent-transaction writes to non-DuckDB tables +# +DROP DATABASE IF EXISTS cross_engine_ryow; +CREATE DATABASE cross_engine_ryow CHARACTER SET utf8mb4; +USE cross_engine_ryow; +CREATE TABLE t_duck (id INT PRIMARY KEY, val VARCHAR(50)) ENGINE=DuckDB; +CREATE TABLE t_inno (id INT PRIMARY KEY, name VARCHAR(50), score INT) ENGINE=InnoDB; +INSERT INTO t_duck VALUES (1,'alpha'),(2,'beta'),(3,'gamma'),(4,'delta'),(5,'epsilon'),(6,'zeta'); +INSERT INTO t_inno VALUES (1,'Alice',90),(2,'Bob',80),(3,'Carol',70),(4,'Dave',60),(5,'Eve',50); + +# (1) RYOW disabled (default): uncommitted writes are NOT visible + +SET SESSION duckdb_cross_engine_ryow=0; +BEGIN; +UPDATE t_inno SET score=999 WHERE id=2; +INSERT INTO t_inno VALUES (6,'Frank',100); +SELECT d.id, i.name, i.score FROM t_duck d JOIN t_inno i ON d.id=i.id; +id name score +1 Alice 90 +2 Bob 80 +3 Carol 70 +4 Dave 60 +5 Eve 50 +ROLLBACK; + +# (2) RYOW enabled: uncommitted writes ARE visible (read own writes) + +SET SESSION duckdb_cross_engine_ryow=1; +BEGIN; +UPDATE t_inno SET score=999 WHERE id=2; +INSERT INTO t_inno VALUES (6,'Frank',100); +SELECT d.id, i.name, i.score FROM t_duck d JOIN t_inno i ON d.id=i.id; +id name score +1 Alice 90 +2 Bob 999 +3 Carol 70 +4 Dave 60 +5 Eve 50 +6 Frank 100 +ROLLBACK; + +# (3) After rollback, committed state is intact + +SELECT d.id, i.name, i.score FROM t_duck d JOIN t_inno i ON d.id=i.id; +id name score +1 Alice 90 +2 Bob 80 +3 Carol 70 +4 Dave 60 +5 Eve 50 + +# (4) RYOW enabled + large external table (rows > DuckDB vector size): +# the direct handler scan spans multiple chunks and must still expose +# the parent transaction's uncommitted write on re-read. + +CREATE TABLE t_inno_big (id INT PRIMARY KEY, score INT) ENGINE=InnoDB; +INSERT INTO t_inno_big SELECT seq, seq FROM seq_1_to_3000; +SET SESSION duckdb_cross_engine_ryow=1; +BEGIN; +UPDATE t_inno_big SET score=999 WHERE id=1; +SELECT i.score FROM t_duck d JOIN t_inno_big i ON d.id=i.id WHERE i.id=1; +score +999 +ROLLBACK; +DROP TABLE t_inno_big; + +# Cleanup + +SET SESSION duckdb_cross_engine_ryow=DEFAULT; +DROP TABLE t_duck; +DROP TABLE t_inno; +DROP DATABASE cross_engine_ryow; diff --git a/storage/duckdb/mysql-test/duckdb/t/cross_engine_ryow.test b/storage/duckdb/mysql-test/duckdb/t/cross_engine_ryow.test new file mode 100644 index 0000000000000..2936b1dd9d2dc --- /dev/null +++ b/storage/duckdb/mysql-test/duckdb/t/cross_engine_ryow.test @@ -0,0 +1,66 @@ +--source include/have_sequence.inc + +--echo # +--echo # Cross-engine RYOW: duckdb_cross_engine_ryow toggles visibility of +--echo # uncommitted parent-transaction writes to non-DuckDB tables +--echo # + +--disable_warnings +DROP DATABASE IF EXISTS cross_engine_ryow; +--enable_warnings +CREATE DATABASE cross_engine_ryow CHARACTER SET utf8mb4; +USE cross_engine_ryow; +CREATE TABLE t_duck (id INT PRIMARY KEY, val VARCHAR(50)) ENGINE=DuckDB; +CREATE TABLE t_inno (id INT PRIMARY KEY, name VARCHAR(50), score INT) ENGINE=InnoDB; +INSERT INTO t_duck VALUES (1,'alpha'),(2,'beta'),(3,'gamma'),(4,'delta'),(5,'epsilon'),(6,'zeta'); +INSERT INTO t_inno VALUES (1,'Alice',90),(2,'Bob',80),(3,'Carol',70),(4,'Dave',60),(5,'Eve',50); + +--echo +--echo # (1) RYOW disabled (default): uncommitted writes are NOT visible +--echo +SET SESSION duckdb_cross_engine_ryow=0; +BEGIN; +UPDATE t_inno SET score=999 WHERE id=2; +INSERT INTO t_inno VALUES (6,'Frank',100); +--sorted_result +SELECT d.id, i.name, i.score FROM t_duck d JOIN t_inno i ON d.id=i.id; +ROLLBACK; + +--echo +--echo # (2) RYOW enabled: uncommitted writes ARE visible (read own writes) +--echo +SET SESSION duckdb_cross_engine_ryow=1; +BEGIN; +UPDATE t_inno SET score=999 WHERE id=2; +INSERT INTO t_inno VALUES (6,'Frank',100); +--sorted_result +SELECT d.id, i.name, i.score FROM t_duck d JOIN t_inno i ON d.id=i.id; +ROLLBACK; + +--echo +--echo # (3) After rollback, committed state is intact +--echo +--sorted_result +SELECT d.id, i.name, i.score FROM t_duck d JOIN t_inno i ON d.id=i.id; + +--echo +--echo # (4) RYOW enabled + large external table (rows > DuckDB vector size): +--echo # the direct handler scan spans multiple chunks and must still expose +--echo # the parent transaction's uncommitted write on re-read. +--echo +CREATE TABLE t_inno_big (id INT PRIMARY KEY, score INT) ENGINE=InnoDB; +INSERT INTO t_inno_big SELECT seq, seq FROM seq_1_to_3000; +SET SESSION duckdb_cross_engine_ryow=1; +BEGIN; +UPDATE t_inno_big SET score=999 WHERE id=1; +SELECT i.score FROM t_duck d JOIN t_inno_big i ON d.id=i.id WHERE i.id=1; +ROLLBACK; +DROP TABLE t_inno_big; + +--echo +--echo # Cleanup +--echo +SET SESSION duckdb_cross_engine_ryow=DEFAULT; +DROP TABLE t_duck; +DROP TABLE t_inno; +DROP DATABASE cross_engine_ryow; diff --git a/storage/duckdb/runtime/cross_engine_scan.cc b/storage/duckdb/runtime/cross_engine_scan.cc index 3e28dce559eed..d0f4c9058b38a 100644 --- a/storage/duckdb/runtime/cross_engine_scan.cc +++ b/storage/duckdb/runtime/cross_engine_scan.cc @@ -54,6 +54,7 @@ static thread_local std::unordered_map tls_external_tables; static thread_local std::unordered_map tls_external_where; +static thread_local bool tls_cross_engine_ryow= false; void register_external_table(const std::string &name, TABLE *table) { @@ -78,6 +79,7 @@ void clear_external_tables() { tls_external_tables.clear(); tls_external_where.clear(); + tls_cross_engine_ryow= false; } TABLE *find_external_table(const std::string &name) @@ -88,6 +90,10 @@ TABLE *find_external_table(const std::string &name) return nullptr; } +void register_cross_engine_ryow(bool enabled) { tls_cross_engine_ryow= enabled; } + +bool cross_engine_ryow_enabled() { return tls_cross_engine_ryow; } + /* ---------------------------------------------------------------- MariaDB Field → DuckDB LogicalType mapping ---------------------------------------------------------------- */ @@ -239,6 +245,7 @@ struct MdbScanBindData : duckdb::FunctionData std::string table_key; std::string where_sql; TABLE *table= nullptr; + bool direct= false; duckdb::unique_ptr Copy() const override { @@ -246,6 +253,7 @@ struct MdbScanBindData : duckdb::FunctionData copy->table_key= table_key; copy->where_sql= where_sql; copy->table= table; + copy->direct= direct; return duckdb::unique_ptr(std::move(copy)); } @@ -264,6 +272,7 @@ struct MdbScanGlobalState : duckdb::GlobalTableFunctionState duckdb::vector column_ids; std::string table_key; std::string where_sql; + bool direct= false; /* Fiber-based scan for predicate pushdown */ std::unique_ptr fiber; @@ -295,6 +304,7 @@ mdb_scan_bind(duckdb::ClientContext &context, data->table_key= key; data->where_sql= find_external_where(key); data->table= tbl; + data->direct= cross_engine_ryow_enabled(); return duckdb::unique_ptr(std::move(data)); } @@ -308,6 +318,7 @@ mdb_scan_init_global(duckdb::ClientContext &context, state->column_ids= input.column_ids; state->table_key= bind_data.table_key; state->where_sql= bind_data.where_sql; + state->direct= bind_data.direct; if ((myduck::duckdb_log_options & LOG_DUCKDB_QUERY) && input.filters) { @@ -365,7 +376,9 @@ static void mdb_scan_function(duckdb::ClientContext &context, if (!state.scan_started) { const std::string &where_sql= state.where_sql; - if (!where_sql.empty() || true) /* Always use fiber for MVP */ + /* RYOW mode reads via the direct handler scan below (parent trx); + otherwise use a fiber running a synthetic SELECT for pushdown. */ + if (!state.direct) { duckdb::vector col_types; uint nfields= tbl->s->fields; @@ -520,14 +533,17 @@ duckdb::unique_ptr mariadb_replacement_scan( children.push_back(duckdb::make_uniq( duckdb::Value(input.table_name))); + const char *func_name= + cross_engine_ryow_enabled() ? "_mdb_scan_direct" : "_mdb_scan"; + ref->function= duckdb::make_uniq( - "_mdb_scan", std::move(children)); + func_name, std::move(children)); ref->alias= input.table_name; if (myduck::duckdb_log_options & LOG_DUCKDB_QUERY) sql_print_information( - "DuckDB cross-engine: replacement scan redirected '%s' to _mdb_scan", - input.table_name.c_str()); + "DuckDB cross-engine: replacement scan redirected '%s' to %s", + input.table_name.c_str(), func_name); return duckdb::unique_ptr(std::move(ref)); } @@ -549,11 +565,22 @@ void register_cross_engine_scan(duckdb::DatabaseInstance &db) auto transaction= duckdb::CatalogTransaction::GetSystemTransaction(db); catalog.CreateFunction(transaction, info); + /* RYOW variant: reads via the parent transaction's handler, so it cannot + honor pushed-down filters. Disable filter_pushdown so DuckDB keeps the + Filter operator above the scan and applies predicates itself. */ + duckdb::TableFunction mdb_scan_direct( + "_mdb_scan_direct", {duckdb::LogicalType::VARCHAR}, mdb_scan_function, + mdb_scan_bind, mdb_scan_init_global); + mdb_scan_direct.projection_pushdown= true; + mdb_scan_direct.filter_pushdown= false; + duckdb::CreateTableFunctionInfo info_direct(std::move(mdb_scan_direct)); + catalog.CreateFunction(transaction, info_direct); + auto &config= duckdb::DBConfig::GetConfig(db); config.replacement_scans.emplace_back(mariadb_replacement_scan); sql_print_information("DuckDB: cross-engine scan registered " - "(_mdb_scan + replacement scan)"); + "(_mdb_scan + _mdb_scan_direct + replacement scan)"); } } /* namespace myduck */ diff --git a/storage/duckdb/runtime/cross_engine_scan.h b/storage/duckdb/runtime/cross_engine_scan.h index 0de92ec92f85f..6932bd3ab15cc 100644 --- a/storage/duckdb/runtime/cross_engine_scan.h +++ b/storage/duckdb/runtime/cross_engine_scan.h @@ -57,6 +57,15 @@ std::string find_external_where(const std::string &name); void clear_external_tables(); TABLE *find_external_table(const std::string &name); +/** + Per-query Read-Your-Own-Writes toggle for cross-engine scans. + When enabled, external tables are read via a direct handler scan in the + parent transaction (RYOW), instead of a fiber running a synthetic SELECT + in a separate background transaction. Set once per query in init_scan. +*/ +void register_cross_engine_ryow(bool enabled); +bool cross_engine_ryow_enabled(); + /** DuckDB replacement scan callback. When DuckDB cannot find a table in its catalog, this callback checks the