mirror of
https://github.com/MariaDB/server.git
synced 2026-10-23 18:49:29 +03:00
fix: DuckDB now optionally read same transactions writes applied to other engines tables.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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's transaction opens its own REPEATABLE READ view; the parent's uncommitted writes are **not** visible) | **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
|
||||
|
||||
```
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
{
|
||||
|
||||
59
storage/duckdb/mysql-test/duckdb/r/cross_engine_ryow.result
Normal file
59
storage/duckdb/mysql-test/duckdb/r/cross_engine_ryow.result
Normal file
@@ -0,0 +1,59 @@
|
||||
#
|
||||
# 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
|
||||
|
||||
# Cleanup
|
||||
|
||||
SET SESSION duckdb_cross_engine_ryow=DEFAULT;
|
||||
DROP TABLE t_duck;
|
||||
DROP TABLE t_inno;
|
||||
DROP DATABASE cross_engine_ryow;
|
||||
50
storage/duckdb/mysql-test/duckdb/t/cross_engine_ryow.test
Normal file
50
storage/duckdb/mysql-test/duckdb/t/cross_engine_ryow.test
Normal file
@@ -0,0 +1,50 @@
|
||||
--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 # Cleanup
|
||||
--echo
|
||||
SET SESSION duckdb_cross_engine_ryow=DEFAULT;
|
||||
DROP TABLE t_duck;
|
||||
DROP TABLE t_inno;
|
||||
DROP DATABASE cross_engine_ryow;
|
||||
@@ -54,6 +54,7 @@ static thread_local std::unordered_map<std::string, TABLE *>
|
||||
tls_external_tables;
|
||||
static thread_local std::unordered_map<std::string, std::string>
|
||||
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<duckdb::FunctionData> 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<duckdb::FunctionData>(std::move(copy));
|
||||
}
|
||||
|
||||
@@ -264,6 +272,7 @@ struct MdbScanGlobalState : duckdb::GlobalTableFunctionState
|
||||
duckdb::vector<duckdb::idx_t> column_ids;
|
||||
std::string table_key;
|
||||
std::string where_sql;
|
||||
bool direct= false;
|
||||
|
||||
/* Fiber-based scan for predicate pushdown */
|
||||
std::unique_ptr<FiberScanState> 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<duckdb::FunctionData>(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<duckdb::LogicalType> col_types;
|
||||
uint nfields= tbl->s->fields;
|
||||
@@ -520,14 +533,17 @@ duckdb::unique_ptr<duckdb::TableRef> mariadb_replacement_scan(
|
||||
children.push_back(duckdb::make_uniq<duckdb::ConstantExpression>(
|
||||
duckdb::Value(input.table_name)));
|
||||
|
||||
const char *func_name=
|
||||
cross_engine_ryow_enabled() ? "_mdb_scan_direct" : "_mdb_scan";
|
||||
|
||||
ref->function= duckdb::make_uniq<duckdb::FunctionExpression>(
|
||||
"_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<duckdb::TableRef>(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 */
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user