RYOW for DuckDB within an explicit txn

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

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
 
 ```

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,

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)
       {

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;

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;

storage/duckdb/runtime/cross_engine_scan.cc

@@ -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,13 +245,15 @@ 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
   {
     auto copy= duckdb::make_uniq<MdbScanBindData>();
     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 */