Bug#35703114 Assertion `m_psi_batch_mode == PSI_BATCH_MODE_NONE' failed.

The fix for bug#34940000 "Hash join execution may be inefficient if
probe input is empty" caused a regression. An assert failure occurs
under the following circumstances:
- We execute a hash join where we start reading from the probe input
  e.g. a left join.
- There is an error when reading that first row, e.g. because there
  is a filter on the probe input where we get a string conversion error.

In this situation, we stop executing the join, but do not turn off PFS
batch mode for the probe iterator. (PFS batch mode is an optimization
for queries involving the performance schema.) This triggers the
assert. This commit fixes this bug by ensuring that we stop PFS batch
mode in this situation.

Change-Id: I49d03d1896f3cff7aa01058e194be2afb89a0f81

Author: Jan Wedvik

mysql-test/include/hash_join.inc

@@ -2713,3 +2713,33 @@ EXPLAIN ANALYZE FORMAT=tree SELECT * FROM t1 x1 WHERE x1.a < 0 AND x1.a NOT IN
   (SELECT a FROM t1 x2 WHERE x2.b = x1.b);
 
 DROP TABLE t1;
+
+--echo #
+--echo # Bug#35703114 Assertion `m_psi_batch_mode == PSI_BATCH_MODE_NONE'
+--echo # failed.
+--echo #
+
+CREATE TABLE t1(i1 int primary key, i2 int, c varchar(10));
+
+INSERT INTO t1 VALUES (1,1,'a'), (2,2,'a'), (3,3,'a'), (4,4,'a'), (5,5,'a');
+
+ANALYZE TABLE t1;
+
+# To reproduce the bug we need a query with these properties:
+# - It should be planned as a hash join.
+# - We should start reading from the probe input (i.e. outer input).
+#   Therefore we use a left join.
+# - There should be a filter on the probe input.
+# - Evaluating this filter should give an error for the first row.
+#   Therefore we use an invalid Unicode code point.
+let $query =
+    SELECT * FROM t1 x1 LEFT JOIN t1 x2 ON x1.i2=x2.i2
+    WHERE x1.c <= CHAR(0xd83f);
+
+--replace_regex $elide_costs_and_rows
+--eval EXPLAIN FORMAT=TREE $query
+
+--error ER_CANNOT_CONVERT_STRING
+--eval $query
+
+DROP TABLE t1;

mysql-test/r/hash_join.result

@@ -4307,3 +4307,25 @@ EXPLAIN
 Warnings:
 Note	1276	Field or reference 'test.x1.b' of SELECT #2 was resolved in SELECT #1
 DROP TABLE t1;
+#
+# Bug#35703114 Assertion `m_psi_batch_mode == PSI_BATCH_MODE_NONE'
+# failed.
+#
+CREATE TABLE t1(i1 int primary key, i2 int, c varchar(10));
+INSERT INTO t1 VALUES (1,1,'a'), (2,2,'a'), (3,3,'a'), (4,4,'a'), (5,5,'a');
+ANALYZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	OK
+EXPLAIN FORMAT=TREE SELECT * FROM t1 x1 LEFT JOIN t1 x2 ON x1.i2=x2.i2
+WHERE x1.c <= CHAR(0xd83f);
+EXPLAIN
+-> Left hash join (x2.i2 = x1.i2)  (...)
+    -> Filter: (x1.c <= <cache>(char(0xd83f)))  (...)
+        -> Table scan on x1  (...)
+    -> Hash
+        -> Table scan on x2  (...)
+
+SELECT * FROM t1 x1 LEFT JOIN t1 x2 ON x1.i2=x2.i2
+WHERE x1.c <= CHAR(0xd83f);
+ERROR HY000: Cannot convert string '\xD8?' from binary to utf8mb4
+DROP TABLE t1;

mysql-test/r/hash_join_hypergraph.result

@@ -4272,6 +4272,28 @@ Warnings:
 Note	1276	Field or reference 'test.x1.b' of SELECT #2 was resolved in SELECT #1
 DROP TABLE t1;
 #
+# Bug#35703114 Assertion `m_psi_batch_mode == PSI_BATCH_MODE_NONE'
+# failed.
+#
+CREATE TABLE t1(i1 int primary key, i2 int, c varchar(10));
+INSERT INTO t1 VALUES (1,1,'a'), (2,2,'a'), (3,3,'a'), (4,4,'a'), (5,5,'a');
+ANALYZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	analyze	status	OK
+EXPLAIN FORMAT=TREE SELECT * FROM t1 x1 LEFT JOIN t1 x2 ON x1.i2=x2.i2
+WHERE x1.c <= CHAR(0xd83f);
+EXPLAIN
+-> Left hash join (x1.i2 = x2.i2)  (...)
+    -> Filter: (x1.c <= <cache>(char(0xd83f)))  (...)
+        -> Table scan on x1  (...)
+    -> Hash
+        -> Table scan on x2  (...)
+
+SELECT * FROM t1 x1 LEFT JOIN t1 x2 ON x1.i2=x2.i2
+WHERE x1.c <= CHAR(0xd83f);
+ERROR HY000: Cannot convert string '\xD8?' from binary to utf8mb4
+DROP TABLE t1;
+#
 # Bug#34940000 Hash join execution may be ineficcient if probe input is empty
 #
 CREATE TABLE t1(

sql/iterators/hash_join_iterator.cc

@@ -153,6 +153,62 @@ bool HashJoinIterator::InitProbeIterator() {
   return false;
 }
 
+bool HashJoinIterator::ReadFirstProbeRow() {
+  assert(m_first_input == HashJoinInput::kProbe);
+  m_state = State::READING_ROW_FROM_PROBE_ITERATOR;
+
+  if (InitProbeIterator()) {
+    return true;
+  }
+
+  const int result = m_probe_input->Read();
+  if (result == 1) {
+    return true;
+
+  } else if (result == -1) {
+    m_state = State::END_OF_ROWS;
+    return false;
+
+  } else {
+    assert(result == 0);
+    m_probe_row_read = true;
+    // Prepare to read the build input into the hash map.
+    PrepareForRequestRowId(m_build_input_tables.tables(),
+                           m_tables_to_get_rowid_for);
+
+    return false;
+  }
+}
+
+bool HashJoinIterator::InitHashTable() {
+  if (BuildHashTable()) {
+    return true;
+  }
+
+  if (m_hash_table_generation != nullptr) {
+    m_last_hash_table_generation = *m_hash_table_generation;
+  }
+
+  if (m_state == State::END_OF_ROWS) {
+    // BuildHashTable() decided that the join is done (the build input is
+    // empty, and we are in an inner-/semijoin. Anti-/outer join must output
+    // NULL-complemented rows from the probe input).
+    return false;
+  }
+
+  if (m_join_type == JoinType::ANTI && m_join_conditions.empty() &&
+      m_extra_condition == nullptr && !m_row_buffer.empty()) {
+    // For degenerate antijoins, we know we will never output anything
+    // if there's anything in the hash table, so we can end right away.
+    // (We also don't need to read more than one row, but
+    // CreateHashJoinAccessPath() has already added a LIMIT 1 for us
+    // in this case.)
+    m_state = State::END_OF_ROWS;
+  }
+
+  return false;
+}
+
 bool HashJoinIterator::Init() {
   // If we are entirely in-memory and the JOIN we are part of hasn't been
   // asked to clear its hash tables since last time, we can reuse the table
@@ -243,73 +299,35 @@ bool HashJoinIterator::Init() {
                          m_tables_to_get_rowid_for);
 
   if (m_first_input == HashJoinInput::kProbe) {
-    m_state = State::READING_ROW_FROM_PROBE_ITERATOR;
-
-    if (InitProbeIterator()) {
-      return true;
-    }
-
-    const int result = m_probe_input->Read();
-    if (result == 1) {
-      assert(thd()->is_error() ||
-             thd()->killed);  // my_error should have been called.
-      return true;
-    } else if (result == -1) {
-      m_probe_input->EndPSIBatchModeIfStarted();
-      m_state = State::END_OF_ROWS;
-      return false;
-    } else {
-      assert(result == 0);
-      m_probe_row_read = true;
-      // Prepare to read the build input into the hash map.
-      PrepareForRequestRowId(m_build_input_tables.tables(),
-                             m_tables_to_get_rowid_for);
-      if (m_build_input->Init()) {
-        assert(thd()->is_error() ||
-               thd()->killed);  // my_error should have been called.
+    const bool error = [&]() {
+      if (ReadFirstProbeRow()) {
         return true;
+      } else if (m_state == State::END_OF_ROWS) {
+        return false;
+      } else {
+        return m_build_input->Init() || InitHashTable();
       }
-    }
-  }
+    }();
 
-  // Build the hash table
-  if (BuildHashTable()) {
-    assert(thd()->is_error() ||
+    assert(!error || thd()->is_error() ||
            thd()->killed);  // my_error should have been called.
-    if (m_first_input == HashJoinInput::kProbe) {
-      m_probe_input->EndPSIBatchModeIfStarted();
-    }
-    return true;
-  }
-  if (m_hash_table_generation != nullptr) {
-    m_last_hash_table_generation = *m_hash_table_generation;
-  }
 
-  if (m_state == State::END_OF_ROWS) {
-    // BuildHashTable() decided that the join is done (the build input is
-    // empty, and we are in an inner-/semijoin. Anti-/outer join must output
-    // NULL-complemented rows from the probe input).
-    if (m_first_input == HashJoinInput::kProbe) {
+    if (m_state == State::END_OF_ROWS || error) {
       m_probe_input->EndPSIBatchModeIfStarted();
     }
-    return false;
-  }
 
-  if (m_join_type == JoinType::ANTI && m_join_conditions.empty() &&
-      m_extra_condition == nullptr && !m_row_buffer.empty()) {
-    // For degenerate antijoins, we know we will never output anything
-    // if there's anything in the hash table, so we can end right away.
-    // (We also don't need to read more than one row, but
-    // CreateHashJoinAccessPath() has already added a LIMIT 1 for us
-    // in this case.)
-    if (m_first_input == HashJoinInput::kProbe) {
-      m_probe_input->EndPSIBatchModeIfStarted();
+    return error;
+
+  } else {  // Start with 'build' input.
+    if (InitHashTable()) {
+      assert(thd()->is_error() ||
+             thd()->killed);  // my_error should have been called.
+
+      return true;
     }
-    m_state = State::END_OF_ROWS;
-    return false;
-  }
 
-  return m_first_input == HashJoinInput::kProbe ? false : InitProbeIterator();
+    return m_state == State::END_OF_ROWS ? false : InitProbeIterator();
+  }
 }
 
 // Construct a join key from a list of join conditions, where the join key from

sql/iterators/hash_join_iterator.h

@@ -677,6 +677,16 @@ class HashJoinIterator final : public RowIterator {
   /// that input was empty. If so, we should process that row before reading
   /// another.
   bool m_probe_row_read{false};
+
+  /// Helper function for Init(). Read the first row from m_probe_input.
+  /// @returns 'true' if there was an error.
+  bool ReadFirstProbeRow();
+
+  /// Helper function for Init(). Build the hash table and check for empty query
+  /// results (empty build input or non-empty build input in case of degenerate
+  /// antijoin.)
+  /// @returns 'true' in case of error.
+  bool InitHashTable();
 };
 
 #endif  // SQL_ITERATORS_HASH_JOIN_ITERATOR_H_