WL#13543 InnoDB: Provide partitioning information to RAPID during load

To support partitioned table in change propagation secondary engine requires
InnoDB to provide partitioning information during table offload.

As part of the offload process the parallel read threads spawned by InnoDB
converts the record from the InnoDB to MySQL format and stacks up the records
in a buffer and sends the buffer to RAPID via a callback.

As part of this worklog,

a) the callback provides the partition ID information of the partition it is
working with
b) all records that are present in the buffer sent via the callback will always
belong to the same partition

RB: 23739
Reviewed-by: Mayank Prasad <mayank.prasad@oracle.com>
Reviewed-by: Sunny Bains <Sunny.Bains@oracle.com>

Author: Darshan M N

share/messages_to_error_log.txt

@@ -10824,6 +10824,9 @@ ER_IB_ERR_PAGE_ARCH_INVALID_FORMAT
 ER_INVALID_XPLUGIN_SOCKET_SAME_AS_SERVER
   eng "X Plugins UNIX socket must use different file than MySQL server. X Plugin won't be accessible through UNIX socket"
 
+ER_INNODB_UNABLE_TO_ACQUIRE_DD_OBJECT
+  eng "%s"
+
 # DO NOT add server-to-client messages here;
 # they go in messages_to_clients.txt
 # in the same directory as this file.

sql/handler.h

@@ -4420,14 +4420,17 @@ class handler {
   /**
     This callback is called by each parallel load thread when processing
     of rows is required for the adapter scan.
-    @param[in] cookie    The cookie for this thread
-    @param[in] nrows     The nrows that are available
-    @param[in] rowdata   The mysql-in-memory row data buffer. This is a memory
-                         buffer for nrows records. The length of each record
-                         is fixed and communicated via Load_init_cbk
+    @param[in] cookie       The cookie for this thread
+    @param[in] nrows        The nrows that are available
+    @param[in] rowdata      The mysql-in-memory row data buffer. This is a
+    memory buffer for nrows records. The length of each record is fixed and
+    communicated via Load_init_cbk
+    @param[in] partition_id Partition id if it's a partitioned table, else
+    std::numeric_limits<uint64_t>::max()
     @returns true if there is an error, false otherwise.
   */
-  using Load_cbk = std::function<bool(void *cookie, uint nrows, void *rowdata)>;
+  using Load_cbk = std::function<bool(void *cookie, uint nrows, void *rowdata,
+                                      uint64_t partition_id)>;
 
   /**
     This callback is called by each parallel load thread when processing

storage/innobase/handler/handler0alter.cc

@@ -9766,8 +9766,21 @@ int ha_innopart::parallel_scan_init(void *&scan_ctx, size_t &num_threads) {
 
   trx_assign_read_view(trx);
 
-  const Parallel_reader::Scan_range FULL_SCAN{};
+  auto dd_client = ha_thd()->dd_client();
+  dd::cache::Dictionary_client::Auto_releaser releaser(dd_client);
+  const dd::Table *dd_table = nullptr;
+
+  if (dd_client->acquire(table_share->db.str, table_share->table_name.str,
+                         &dd_table)) {
+    ib::error(ER_INNODB_UNABLE_TO_ACQUIRE_DD_OBJECT)
+        << "InnoDB can't get table object for table"
+        << table_share->table_name.str;
+    return HA_ERR_INTERNAL_ERROR;
+  }
 
+  auto dd_partitions = dd_table->leaf_partitions();
+
+  const Parallel_reader::Scan_range FULL_SCAN{};
   const auto first_used_partition = m_part_info->get_first_used_partition();
 
   for (auto i = first_used_partition; i < m_tot_parts;
@@ -9779,12 +9792,13 @@ int ha_innopart::parallel_scan_init(void *&scan_ctx, size_t &num_threads) {
                   m_prebuilt->table->name.m_name);
 
       UT_DELETE(adapter);
-      return (HA_ERR_NO_SUCH_TABLE);
+      return HA_ERR_NO_SUCH_TABLE;
     }
 
     build_template(true);
 
-    Parallel_reader::Config config(FULL_SCAN, m_prebuilt->table->first_index());
+    Parallel_reader::Config config(FULL_SCAN, m_prebuilt->table->first_index(),
+                                   0, dd_partitions[i]->number());
 
     dberr_t err =
         adapter->add_scan(trx, config, [=](const Parallel_reader::Ctx *ctx) {

storage/innobase/include/row0pread-adapter.h

@@ -1,6 +1,6 @@
 /*****************************************************************************
 
-Copyright (c) 2018, 2019, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2018, 2020, Oracle and/or its affiliates. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License, version 2.0, as published by the
@@ -95,15 +95,18 @@ class Parallel_reader_adapter {
   dberr_t init(size_t thread_id) MY_ATTRIBUTE((warn_unused_result));
 
   /** For pushing any left over rows to the caller.
-  @param[in]  thread_id         ID of the thread.
+  @param[in]  ctx       Parallel read context.
+  @param[in]  thread_id ID of the thread.
   @return DB_SUCCESS or error code. */
-  dberr_t end(size_t thread_id) MY_ATTRIBUTE((warn_unused_result));
+  dberr_t end(Parallel_reader::Ctx *ctx, size_t thread_id)
+      MY_ATTRIBUTE((warn_unused_result));
 
   /** Send a batch of records.
-  @param[in]  thread_id         ID of the thread.
-  @param[in]  n_recs            Number of records to send.
+  @param[in]  ctx       Parallel read context.
+  @param[in]  thread_id ID of the thread.
+  @param[in]  n_recs    Number of records to send.
   @return DB_SUCCESS or error code. */
-  dberr_t send_batch(size_t thread_id, uint64_t n_recs)
+  dberr_t send_batch(const Parallel_reader::Ctx *ctx, uint64_t n_recs)
       MY_ATTRIBUTE((warn_unused_result));
 
   /** Get the number of rows buffered but not sent.

storage/innobase/include/row0pread.h

@@ -116,7 +116,8 @@ class Parallel_reader {
   using Start = std::function<dberr_t(size_t thread_id)>;
 
   /** Callback to finalise callers state. */
-  using Finish = std::function<dberr_t(size_t thread_id)>;
+  using Finish =
+      std::function<dberr_t(Parallel_reader::Ctx *ctx, size_t thread_id)>;
 
   /** Callback to process the rows. */
   using F = std::function<dberr_t(const Ctx *)>;
@@ -150,18 +151,23 @@ class Parallel_reader {
   /** Scan (Scan_ctx) configuration. */
   struct Config {
     /** Constructor.
-    @param[in] scan_range       Range to scan.
-    @param[in] index            Cluster index to scan.
-    @param[in] read_level       Btree level from which records need to be read.
-    @param[in] read_ahead       If true then start read ahead threads.
+    @param[in] scan_range   Range to scan.
+    @param[in] index        Cluster index to scan.
+    @param[in] read_level   Btree level from which records need to be read.
+    @param[in] partition_id Partition id if it the index to be scanned belongs
+    to a partitioned table.
+    @param[in] read_ahead   If true then start read ahead threads.
   */
     Config(const Scan_range &scan_range, dict_index_t *index,
-           size_t read_level = 0, bool read_ahead = false)
+           size_t read_level = 0,
+           size_t partition_id = std::numeric_limits<uint64_t>::max(),
+           bool read_ahead = false)
         : m_scan_range(scan_range),
           m_index(index),
           m_is_compact(dict_table_is_comp(index->table)),
           m_page_size(dict_tf_to_fsp_flags(index->table->flags)),
           m_read_level(read_level),
+          m_partition_id(partition_id),
           m_read_ahead(read_ahead) {}
 
     /** Copy constructor.
@@ -172,6 +178,7 @@ class Parallel_reader {
           m_is_compact(config.m_is_compact),
           m_page_size(config.m_page_size),
           m_read_level(config.m_read_level),
+          m_partition_id(config.m_partition_id),
           m_read_ahead(config.m_read_ahead) {}
 
     /** Range to scan. */
@@ -189,6 +196,10 @@ class Parallel_reader {
     /** Btree level from which records need to be read. */
     size_t m_read_level{0};
 
+    /** Partition id if the index to be scanned belongs to a partitioned table,
+    else std::numeric_limits<uint64_t>::max(). */
+    size_t m_partition_id{std::numeric_limits<uint64_t>::max()};
+
     /** if true then enable separate read ahead threads. */
     bool m_read_ahead{false};
   };
@@ -697,6 +708,13 @@ class Parallel_reader::Ctx {
     return (m_scan_ctx->m_config.m_index);
   }
 
+  /** @return the partition id of the index.
+  @note this is std::numeric_limits<uint64_t>::max() if the index does not
+  belong to a partition. */
+  size_t partition_id() const MY_ATTRIBUTE((warn_unused_result)) {
+    return m_scan_ctx->m_config.m_partition_id;
+  }
+
  private:
   /** Traverse the pages by key order.
   @return DB_SUCCESS or error code. */

storage/innobase/row/row0pread-adapter.cc

@@ -96,8 +96,8 @@ void Parallel_reader_adapter::set(row_prebuilt_t *prebuilt) {
     return (init(thread_id));
   });
 
-  m_parallel_reader.set_finish_callback([=](size_t thread_id) {
-    return (end(thread_id));
+  m_parallel_reader.set_finish_callback([=](Parallel_reader::Ctx *ctx, size_t thread_id) {
+      return end(ctx, thread_id);
   });
   // clang-format on
 
@@ -115,10 +115,13 @@ dberr_t Parallel_reader_adapter::run(void **thread_contexts, Init_fn init_fn,
   return (m_parallel_reader.run());
 }
 
-dberr_t Parallel_reader_adapter::send_batch(size_t thread_id, uint64_t n_recs) {
+dberr_t Parallel_reader_adapter::send_batch(const Parallel_reader::Ctx *ctx,
+                                            uint64_t n_recs) {
   ut_a(n_recs <= m_batch_size);
 
   dberr_t err{DB_SUCCESS};
+  auto thread_id = ctx->m_thread_id;
+  size_t id = ctx->partition_id();
 
   /* Push the row buffer to the caller if we have filled the buffer with
   ADAPTER_SEND_NUM_RECS number of records or it's a start of a new range. */
@@ -131,7 +134,7 @@ dberr_t Parallel_reader_adapter::send_batch(size_t thread_id, uint64_t n_recs) {
 
   const auto p = &buffer[start * m_mysql_row.m_max_len];
 
-  if (m_load_fn(m_thread_contexts[thread_id], static_cast<uint>(n_recs), p)) {
+  if (m_load_fn(m_thread_contexts[thread_id], n_recs, p, id)) {
     err = DB_INTERRUPTED;
     m_parallel_reader.set_error_state(DB_INTERRUPTED);
   }
@@ -175,7 +178,7 @@ dberr_t Parallel_reader_adapter::process_rows(const Parallel_reader::Ctx *ctx) {
 
     /* Start of a new range, send what we have buffered. */
     if (ctx->m_start && n_pending > 0) {
-      err = send_batch(thread_id, n_pending);
+      err = send_batch(ctx, n_pending);
 
       if (err != DB_SUCCESS) {
         if (heap != nullptr) {
@@ -203,7 +206,7 @@ dberr_t Parallel_reader_adapter::process_rows(const Parallel_reader::Ctx *ctx) {
       originated from RAPID threads. */
       err = DB_ERROR;
     } else if (is_buffer_full(thread_id)) {
-      err = send_batch(thread_id, pending(thread_id));
+      err = send_batch(ctx, pending(thread_id));
     }
   } else {
     err = DB_ERROR;
@@ -216,19 +219,20 @@ dberr_t Parallel_reader_adapter::process_rows(const Parallel_reader::Ctx *ctx) {
   return (err);
 }
 
-dberr_t Parallel_reader_adapter::end(size_t thread_id) {
+dberr_t Parallel_reader_adapter::end(Parallel_reader::Ctx *ctx,
+                                     size_t thread_id) {
   ut_a(Counter::get(m_n_sent, thread_id) <= Counter::get(m_n_read, thread_id));
 
   ut_a((Counter::get(m_n_read, thread_id) -
         Counter::get(m_n_sent, thread_id)) <= m_batch_size);
 
   dberr_t err{DB_SUCCESS};
 
-  if (!m_parallel_reader.is_error_set()) {
+  if (!m_parallel_reader.is_error_set() && ctx != nullptr) {
     /* It's possible that we might not have sent the records in the buffer
     when we have reached the end of records and the buffer is not full.
     Send them now. */
-    err = (pending(thread_id) != 0) ? send_batch(thread_id, pending(thread_id))
+    err = (pending(thread_id) != 0) ? send_batch(ctx, pending(thread_id))
                                     : DB_SUCCESS;
   }
 

storage/innobase/row/row0pread-histogram.cc

@@ -1,6 +1,6 @@
 /*****************************************************************************
 
-Copyright (c) 2020, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2019, 2020 Oracle and/or its affiliates. All Rights Reserved.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License, version 2.0, as published by the
@@ -62,7 +62,8 @@ Histogram_sampler::Histogram_sampler(size_t max_threads, int sampling_seed,
 
   m_n_sampled = 0;
 
-  m_parallel_reader.set_finish_callback([&](size_t thread_id) {
+  m_parallel_reader.set_finish_callback([&](Parallel_reader::Ctx *ctx,
+                                            size_t thread_id) {
     DBUG_PRINT("histogram_sampler_buffering_print", ("-> Buffering complete."));
 
     DBUG_LOG("histogram_sampler_buffering_print",

storage/innobase/row/row0pread.cc

@@ -636,6 +636,8 @@ void Parallel_reader::worker(size_t thread_id) {
   constexpr auto FOREVER = OS_SYNC_INFINITE_TIME;
   os_event_wait_time_low(m_event, FOREVER, m_sig_count);
 
+  std::shared_ptr<Parallel_reader::Ctx> last_ctx{nullptr};
+
   for (;;) {
     size_t n_completed = 0;
     int64_t sig_count = os_event_reset(m_event);
@@ -664,6 +666,7 @@ void Parallel_reader::worker(size_t thread_id) {
       }
 
       ++n_completed;
+      last_ctx = std::move(ctx);
     }
 
     if (err != DB_SUCCESS || is_error_set()) {
@@ -696,9 +699,12 @@ void Parallel_reader::worker(size_t thread_id) {
   }
 
   if (m_finish_callback) {
-    dberr_t finish_err = m_finish_callback(thread_id);
+    dberr_t finish_err = m_finish_callback(last_ctx.get(), thread_id);
+
     /* Keep the err status from previous failed operations */
-    if (unlikely(finish_err != DB_SUCCESS)) err = finish_err;
+    if (finish_err != DB_SUCCESS) {
+      err = finish_err;
+    }
   }
 
   ut_a(err != DB_SUCCESS || is_error_set() ||