WL#15730: Lakehouse full query support

 - This provides random scan and parallel scan implementation for
   Lakehouse tables to use MySQL query engine as fallback for queries
   not supported by the Rapid secondary engine.
 - It issues a select query on rapid everytime a scan needs to be
   performed
 - For random scan support (general MySQL queries), a separate thread
   is spawned for processing the results from the Rapid secondary
   engine. This enables pipelining between parsing of results and
   feeding the MySQL query engine with rows. If the MySQL query
   finishes the random scan, the Rapid query is aborted.
 - Implementation of the modules for basic accesses for a lakehouse
   table by processing the next record or a stored record.
 - Performance schema statistics per query and per lakehouse table.
   Show for each query offloaded to lakehouse how many scans it
   performed and any failure message.
   Similarly for each loaded lakehouse table that participates in a
   query offloaded to rapid show the number of started scans,
   successful scans, failed scans and aborted scans.

Change-Id: If54204be287bc689d757dd011f2b5e1bbb124fa4

Author: Darko Makreshanski

sql/handler.h

@@ -2442,6 +2442,16 @@ using compare_secondary_engine_cost_t = bool (*)(THD *thd, const JOIN &join,
 using secondary_engine_modify_access_path_cost_t = bool (*)(
     THD *thd, const JoinHypergraph &hypergraph, AccessPath *access_path);
 
+/**
+  Checks whether the tables used in an explain query are loaded in the secondary
+  engine.
+  @param thd thread context.
+
+  @retval true if there is a table not loaded to the secondary engine, false
+  otherwise
+*/
+using external_engine_explain_check_t = bool (*)(THD *thd);
+
 /**
   Looks up and returns a specific secondary engine query offload or exec
   failure reason as a string given a thread context (representing the query)
@@ -2914,6 +2924,12 @@ struct handlerton {
   /// does not support the hypergraph join optimizer.
   SecondaryEngineFlags secondary_engine_flags;
 
+  /// Pointer to a function that checks if the table is loaded in the
+  /// secondary engine in the case of an explain statement.
+  ///
+  /// @see external_engine_explain_check_t for function signature.
+  external_engine_explain_check_t external_engine_explain_check;
+
   /// Pointer to a function that evaluates the cost of executing an access path
   /// in a secondary storage engine.
   ///

sql/sql_base.cc

@@ -6915,14 +6915,6 @@ bool open_tables_for_query(THD *thd, Table_ref *tables, uint flags) {
                   &prelocking_strategy))
     goto end;
 
-  for (Table_ref *tr = tables; tr != nullptr; tr = tr->next_global) {
-    if (tr->is_external()) {
-      thd->lex->set_execute_only_in_secondary_engine(
-          /*execute_only_in_secondary_engine_param=*/true, EXTERNAL);
-      break;
-    }
-  }
-
   if (open_secondary_engine_tables(thd, flags)) goto end;
 
   return false;

sql/sql_lex.cc

@@ -4344,24 +4344,6 @@ bool LEX::locate_var_assignment(const Name_string &name) {
   return false;
 }
 
-/**
-  Get the reason why a query cannot be executed in the primary engine.
-
-  @returns a text representation of the reason the query is not supported in the
-  primary engine.
-*/
-const char *LEX::get_not_supported_in_primary_reason() {
-  assert(can_execute_only_in_secondary_engine());
-  switch (m_execute_only_in_secondary_engine_reason) {
-    case CUBE:
-      return "CUBE";
-    case EXTERNAL:
-      return "External engine as primary storage engine";
-    default:
-      return "UNDEFINED";
-  }
-}
-
 /**
   Save properties for ORDER clauses so that they can be reconstructed
   for a new optimization of the query block.

sql/sql_lex.h

@@ -3725,7 +3725,7 @@ class LEX_GRANT_AS {
   "execute_only_in_secondary_reasons" retains the explanations for queries that
   cannot be executed using the primary engine.
 */
-enum execute_only_in_secondary_reasons { SUPPORTED_IN_PRIMARY, CUBE, EXTERNAL };
+enum execute_only_in_secondary_reasons { SUPPORTED_IN_PRIMARY, CUBE };
 
 /*
   Some queries can be executed only in using the hypergraph optimizer. The enum
@@ -3873,10 +3873,6 @@ struct LEX : public Query_tables_list {
   /// Leaf table being inserted into (always a base table)
   Table_ref *insert_table_leaf;
 
-  bool has_external_tables() const {
-    return (m_execute_only_in_secondary_engine_reason == EXTERNAL);
-  }
-
   /** SELECT of CREATE VIEW statement */
   LEX_STRING create_view_query_block;
 
@@ -3955,8 +3951,15 @@ struct LEX : public Query_tables_list {
            reason == SUPPORTED_IN_PRIMARY);
   }
 
-  const char *get_not_supported_in_primary_reason();
-
+  const char *get_not_supported_in_primary_reason() {
+    assert(can_execute_only_in_secondary_engine());
+    switch (m_execute_only_in_secondary_engine_reason) {
+      case CUBE:
+        return "CUBE";
+      default:
+        return "UNDEFINED";
+    }
+  }
   bool can_execute_only_in_hypergraph_optimizer() const {
     return m_can_execute_only_in_hypergraph_optimizer;
   }

sql/sql_select.cc

@@ -377,19 +377,6 @@ std::string_view get_secondary_engine_fail_reason(const LEX *lex) {
   return {};
 }
 
-void set_external_engine_fail_reason(const LEX *lex, const char *reason) {
-  if (!lex->thd->is_secondary_engine_forced() && reason != nullptr) {
-    for (Table_ref *ref = lex->query_tables; ref != nullptr;
-         ref = ref->next_global) {
-      if (ref->is_external()) {
-        ref->table->get_primary_handler()->set_external_table_offload_error(
-            reason);
-        break;
-      }
-    }
-  }
-}
-
 std::string_view find_secondary_engine_fail_reason(const LEX *lex) {
   const auto *hton = get_secondary_engine_handlerton(lex);
   if (hton != nullptr &&
@@ -673,6 +660,19 @@ bool Sql_cmd_select::prepare_inner(THD *thd) {
   return false;
 }
 
+bool has_external_table(const LEX *lex) {
+  if (lex->m_sql_cmd == nullptr) {
+    return false;
+  }
+  for (Table_ref *ref = lex->query_tables; ref != nullptr;
+       ref = ref->next_global) {
+    if (ref->is_external()) {
+      return true;
+    }
+  }
+  return false;
+}
+
 bool Sql_cmd_dml::execute(THD *thd) {
   DBUG_TRACE;
 
@@ -909,8 +909,11 @@ static bool retry_with_secondary_engine(THD *thd) {
 
   // Only attempt to use the secondary engine if the estimated cost of the query
   // is higher than the specified cost threshold.
-  if (thd->m_current_query_cost <=
-      thd->variables.secondary_engine_cost_threshold) {
+  // We allow any query to be executed in the secondary_engine when it involves
+  // external tables.
+  if (!has_external_table(thd->lex) &&
+      (thd->m_current_query_cost <=
+       thd->variables.secondary_engine_cost_threshold)) {
     Opt_trace_context *const trace = &thd->opt_trace;
     if (trace->is_started()) {
       const Opt_trace_object wrapper(trace);
@@ -1000,6 +1003,16 @@ bool Sql_cmd_dml::execute_inner(THD *thd) {
   // We know by now that execution will complete (successful or with error)
   lex->set_exec_completed();
   if (lex->is_explain()) {
+    for (Table_ref *ref = lex->query_tables; ref != nullptr;
+         ref = ref->next_global) {
+      if (ref->table != nullptr && ref->table->file != nullptr) {
+        handlerton *hton = ref->table->file->ht;
+        if (hton->external_engine_explain_check != nullptr) {
+          if (hton->external_engine_explain_check(thd)) return true;
+        }
+      }
+    }
+
     if (explain_query(thd, thd, unit)) return true; /* purecov: inspected */
   } else {
     if (unit->execute(thd)) return true;

sql/sql_select.h

@@ -1060,6 +1060,14 @@ void accumulate_statement_cost(const LEX *lex);
 */
 const handlerton *get_secondary_engine_handlerton(const LEX *lex);
 
+/**
+  Checks if any of the tables referenced belong to an external engine.
+  If an external table is found, return true, false otherwise.
+
+  @param lex the statement
+*/
+bool has_external_table(const LEX *lex);
+
 /**
   Sets the reason of failure for the statement to the external engine.
 

sql/table.cc

@@ -7324,13 +7324,17 @@ uint Table_ref::get_hidden_field_count_for_derived() const {
 }
 
 bool Table_ref::is_external() const {
-  if (is_placeholder()) return false;
-  assert(table != nullptr);
-  handler *primary_handler = table->get_primary_handler();
-  assert(primary_handler != nullptr);
-
-  return Overlaps(primary_handler->ht->flags, HTON_SUPPORTS_EXTERNAL_SOURCE) &&
-         primary_handler->get_table_share()->has_secondary_engine();
+  if (m_table_ref_type == TABLE_REF_BASE_TABLE && table != nullptr &&
+      table->file != nullptr) {
+    if (is_placeholder()) return false;
+    handler *primary_handler = table->get_primary_handler();
+    return primary_handler != nullptr &&
+           Overlaps(primary_handler->ht->flags,
+                    HTON_SUPPORTS_EXTERNAL_SOURCE) &&
+           primary_handler->get_table_share() != nullptr &&
+           primary_handler->get_table_share()->has_secondary_engine();
+  }
+  return false;
 }
 
 void LEX_MFA::copy(LEX_MFA *m, MEM_ROOT *alloc) {