5.6.42-84.2

35 files changed, 1778 insertions, 1075 deletions

diff --git a/storage/tokudb/PerconaFT/COPYING.APACHEv2 b/storage/tokudb/PerconaFT/COPYING.APACHEv2
new file mode 100644
index 00000000000..ecbfc770fa9
--- /dev/null
+++ b/storage/tokudb/PerconaFT/COPYING.APACHEv2
@@ -0,0 +1,174 @@
+Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
diff --git a/storage/tokudb/PerconaFT/README.md b/storage/tokudb/PerconaFT/README.md
index ffb646b67af..26333df877e 100644
--- a/storage/tokudb/PerconaFT/README.md
+++ b/storage/tokudb/PerconaFT/README.md
@@ -104,11 +104,14 @@ All source code and test contributions must be provided under a [BSD 2-Clause][b
 License
 -------
 
+Portions of the PerconaFT library (the 'locktree' and 'omt') are available under the Apache version 2 license.
 PerconaFT is available under the GPL version 2, and AGPL version 3.
-See [COPYING.AGPLv3][agpllicense],
+See [COPYING.APACHEv2][apachelicense],
+[COPYING.AGPLv3][agpllicense],
 [COPYING.GPLv2][gpllicense], and
 [PATENTS][patents].
 
+[apachelicense]: http://github.com/Percona/PerconaFT/blob/master/COPYING.APACHEv2
 [agpllicense]: http://github.com/Percona/PerconaFT/blob/master/COPYING.AGPLv3
 [gpllicense]: http://github.com/Percona/PerconaFT/blob/master/COPYING.GPLv2
 [patents]: http://github.com/Percona/PerconaFT/blob/master/PATENTS
diff --git a/storage/tokudb/PerconaFT/ft/txn/txn_manager.h b/storage/tokudb/PerconaFT/ft/txn/txn_manager.h
index 7cdc52c4f43..25fa6032112 100644
--- a/storage/tokudb/PerconaFT/ft/txn/txn_manager.h
+++ b/storage/tokudb/PerconaFT/ft/txn/txn_manager.h
@@ -46,11 +46,11 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 void set_test_txn_sync_callback(void (*) (pthread_t, void*), void*);
 #define toku_test_txn_sync_callback(a) ((test_txn_sync_callback)? test_txn_sync_callback( a,test_txn_sync_callback_extra) : (void) 0)
 
-#if TOKU_DEBUG_TXN_SYNC
+#if defined(TOKU_DEBUG_TXN_SYNC)
 #define toku_debug_txn_sync(a) toku_test_txn_sync_callback(a) 
 #else
 #define toku_debug_txn_sync(a) ((void) 0)
-#endif
+#endif  // defined(TOKU_DEBUG_TXN_SYNC)
 
 typedef struct txn_manager *TXN_MANAGER;
 
diff --git a/storage/tokudb/PerconaFT/locktree/concurrent_tree.cc b/storage/tokudb/PerconaFT/locktree/concurrent_tree.cc
index 9347267db49..e07f32c98fb 100644
--- a/storage/tokudb/PerconaFT/locktree/concurrent_tree.cc
+++ b/storage/tokudb/PerconaFT/locktree/concurrent_tree.cc
@@ -32,6 +32,20 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License. 
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/concurrent_tree.h b/storage/tokudb/PerconaFT/locktree/concurrent_tree.h
index 1eb339b7317..66a7ff176bb 100644
--- a/storage/tokudb/PerconaFT/locktree/concurrent_tree.h
+++ b/storage/tokudb/PerconaFT/locktree/concurrent_tree.h
@@ -32,6 +32,20 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License. 
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/keyrange.cc b/storage/tokudb/PerconaFT/locktree/keyrange.cc
index 8c2a69d4703..2b4b3bbd4fd 100644
--- a/storage/tokudb/PerconaFT/locktree/keyrange.cc
+++ b/storage/tokudb/PerconaFT/locktree/keyrange.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/keyrange.h b/storage/tokudb/PerconaFT/locktree/keyrange.h
index 079ac3d7a80..a454287cbc8 100644
--- a/storage/tokudb/PerconaFT/locktree/keyrange.h
+++ b/storage/tokudb/PerconaFT/locktree/keyrange.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/lock_request.cc b/storage/tokudb/PerconaFT/locktree/lock_request.cc
index 8d49ccf8a1f..3d4d43b9e25 100644
--- a/storage/tokudb/PerconaFT/locktree/lock_request.cc
+++ b/storage/tokudb/PerconaFT/locktree/lock_request.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/lock_request.h b/storage/tokudb/PerconaFT/locktree/lock_request.h
index a8d8cb7785b..91a6ff12b52 100644
--- a/storage/tokudb/PerconaFT/locktree/lock_request.h
+++ b/storage/tokudb/PerconaFT/locktree/lock_request.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/locktree.cc b/storage/tokudb/PerconaFT/locktree/locktree.cc
index 069aae26f66..8ba3f0f00ae 100644
--- a/storage/tokudb/PerconaFT/locktree/locktree.cc
+++ b/storage/tokudb/PerconaFT/locktree/locktree.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/locktree.h b/storage/tokudb/PerconaFT/locktree/locktree.h
index 1ba7a51b124..7006b6fb01d 100644
--- a/storage/tokudb/PerconaFT/locktree/locktree.h
+++ b/storage/tokudb/PerconaFT/locktree/locktree.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/manager.cc b/storage/tokudb/PerconaFT/locktree/manager.cc
index 6bb5c77bf32..21f8dc6cf01 100644
--- a/storage/tokudb/PerconaFT/locktree/manager.cc
+++ b/storage/tokudb/PerconaFT/locktree/manager.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/range_buffer.cc b/storage/tokudb/PerconaFT/locktree/range_buffer.cc
index 3ddfd0faf97..d1f14fc4a52 100644
--- a/storage/tokudb/PerconaFT/locktree/range_buffer.cc
+++ b/storage/tokudb/PerconaFT/locktree/range_buffer.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/range_buffer.h b/storage/tokudb/PerconaFT/locktree/range_buffer.h
index b0e36968e73..811b0f85e69 100644
--- a/storage/tokudb/PerconaFT/locktree/range_buffer.h
+++ b/storage/tokudb/PerconaFT/locktree/range_buffer.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/treenode.cc b/storage/tokudb/PerconaFT/locktree/treenode.cc
index cc3a4969643..0247242f975 100644
--- a/storage/tokudb/PerconaFT/locktree/treenode.cc
+++ b/storage/tokudb/PerconaFT/locktree/treenode.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/treenode.h b/storage/tokudb/PerconaFT/locktree/treenode.h
index 08aad2b6636..981e8b5a9cf 100644
--- a/storage/tokudb/PerconaFT/locktree/treenode.h
+++ b/storage/tokudb/PerconaFT/locktree/treenode.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/txnid_set.cc b/storage/tokudb/PerconaFT/locktree/txnid_set.cc
index 82b59453156..bd4e9723155 100644
--- a/storage/tokudb/PerconaFT/locktree/txnid_set.cc
+++ b/storage/tokudb/PerconaFT/locktree/txnid_set.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/txnid_set.h b/storage/tokudb/PerconaFT/locktree/txnid_set.h
index 109d7f798e4..81fd45b6dde 100644
--- a/storage/tokudb/PerconaFT/locktree/txnid_set.h
+++ b/storage/tokudb/PerconaFT/locktree/txnid_set.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/wfg.cc b/storage/tokudb/PerconaFT/locktree/wfg.cc
index 9a234f50060..26b7a3b5295 100644
--- a/storage/tokudb/PerconaFT/locktree/wfg.cc
+++ b/storage/tokudb/PerconaFT/locktree/wfg.cc
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/locktree/wfg.h b/storage/tokudb/PerconaFT/locktree/wfg.h
index c56886e1362..5c1599592e6 100644
--- a/storage/tokudb/PerconaFT/locktree/wfg.h
+++ b/storage/tokudb/PerconaFT/locktree/wfg.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/portability/toku_instr_mysql.cc b/storage/tokudb/PerconaFT/portability/toku_instr_mysql.cc
index 6f69c3c31b9..b5305ffaff4 100644
--- a/storage/tokudb/PerconaFT/portability/toku_instr_mysql.cc
+++ b/storage/tokudb/PerconaFT/portability/toku_instr_mysql.cc
@@ -184,9 +184,9 @@ void toku_instr_file_io_end(toku_io_instrumentation &io_instr, ssize_t count) {
 
 void toku_instr_mutex_init(const toku_instr_key &key, toku_mutex_t &mutex) {
     mutex.psi_mutex = PSI_MUTEX_CALL(init_mutex)(key.id(), &mutex.pmutex);
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     mutex.instr_key_id = key.id();
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 }
 
 void toku_instr_mutex_destroy(PSI_mutex *&mutex_instr) {
@@ -242,9 +242,9 @@ void toku_instr_mutex_unlock(PSI_mutex *mutex_instr) {
 
 void toku_instr_cond_init(const toku_instr_key &key, toku_cond_t &cond) {
     cond.psi_cond = PSI_COND_CALL(init_cond)(key.id(), &cond.pcond);
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     cond.instr_key_id = key.id();
-#endif
+#endif  //   // defined(TOKU_PTHREAD_DEBUG)
 }
 
 void toku_instr_cond_destroy(PSI_cond *&cond_instr) {
@@ -295,9 +295,9 @@ void toku_instr_cond_broadcast(const toku_cond_t &cond) {
 void toku_instr_rwlock_init(const toku_instr_key &key,
                             toku_pthread_rwlock_t &rwlock) {
     rwlock.psi_rwlock = PSI_RWLOCK_CALL(init_rwlock)(key.id(), &rwlock.rwlock);
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     rwlock.instr_key_id = key.id();
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 }
 
 void toku_instr_rwlock_destroy(PSI_rwlock *&rwlock_instr) {
diff --git a/storage/tokudb/PerconaFT/portability/toku_instr_mysql.h b/storage/tokudb/PerconaFT/portability/toku_instr_mysql.h
index d6b0ed35ce9..695624acd6d 100644
--- a/storage/tokudb/PerconaFT/portability/toku_instr_mysql.h
+++ b/storage/tokudb/PerconaFT/portability/toku_instr_mysql.h
@@ -12,8 +12,15 @@
 // undefine them here to avoid compilation errors.
 #undef __STDC_FORMAT_MACROS
 #undef __STDC_LIMIT_MACROS
-#include <mysql/psi/mysql_file.h>    // PSI_file
-#include <mysql/psi/mysql_thread.h>  // PSI_mutex
+#include "mysql/psi/mysql_file.h"    // PSI_file
+#include "mysql/psi/mysql_thread.h"  // PSI_mutex
+#include "mysql/psi/mysql_stage.h"   // PSI_stage
+
+#if (MYSQL_VERSION_ID >= 80000)
+#include "mysql/psi/mysql_cond.h"
+#include "mysql/psi/mysql_mutex.h"
+#include "mysql/psi/mysql_rwlock.h"
+#endif  //  (MYSQL_VERSION_ID >= nn)
 
 #ifndef HAVE_PSI_MUTEX_INTERFACE
 #error HAVE_PSI_MUTEX_INTERFACE required
diff --git a/storage/tokudb/PerconaFT/portability/toku_pthread.h b/storage/tokudb/PerconaFT/portability/toku_pthread.h
index e3bd3bce598..da956097d05 100644
--- a/storage/tokudb/PerconaFT/portability/toku_pthread.h
+++ b/storage/tokudb/PerconaFT/portability/toku_pthread.h
@@ -64,23 +64,23 @@ struct toku_mutex_t {
     pthread_mutex_t pmutex;
     struct PSI_mutex
         *psi_mutex; /* The performance schema instrumentation hook */
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     pthread_t owner;  // = pthread_self(); // for debugging
     bool locked;
     bool valid;
     pfs_key_t instr_key_id;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 };
 
 struct toku_cond_t {
     pthread_cond_t pcond;
     struct PSI_cond *psi_cond;
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     pfs_key_t instr_key_id;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 };
 
-#ifdef TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
 #define TOKU_COND_INITIALIZER                                   \
     {                                                           \
         .pcond = PTHREAD_COND_INITIALIZER, .psi_cond = nullptr, \
@@ -89,14 +89,14 @@ struct toku_cond_t {
 #else
 #define TOKU_COND_INITIALIZER \
     { .pcond = PTHREAD_COND_INITIALIZER, .psi_cond = nullptr }
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 
 struct toku_pthread_rwlock_t {
     pthread_rwlock_t rwlock;
     struct PSI_rwlock *psi_rwlock;
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     pfs_key_t instr_key_id;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 };
 
 typedef struct toku_mutex_aligned {
@@ -117,7 +117,7 @@ typedef struct toku_mutex_aligned {
 #define ZERO_MUTEX_INITIALIZER \
     {}
 
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
 #define TOKU_MUTEX_INITIALIZER                                                 \
     {                                                                          \
         .pmutex = PTHREAD_MUTEX_INITIALIZER, .psi_mutex = nullptr, .owner = 0, \
@@ -126,12 +126,12 @@ typedef struct toku_mutex_aligned {
 #else
 #define TOKU_MUTEX_INITIALIZER \
     { .pmutex = PTHREAD_MUTEX_INITIALIZER, .psi_mutex = nullptr }
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 
 // Darwin doesn't provide adaptive mutexes
 #if defined(__APPLE__)
 #define TOKU_MUTEX_ADAPTIVE PTHREAD_MUTEX_DEFAULT
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
 #define TOKU_ADAPTIVE_MUTEX_INITIALIZER                                        \
     {                                                                          \
         .pmutex = PTHREAD_MUTEX_INITIALIZER, .psi_mutex = nullptr, .owner = 0, \
@@ -140,10 +140,10 @@ typedef struct toku_mutex_aligned {
 #else
 #define TOKU_ADAPTIVE_MUTEX_INITIALIZER \
     { .pmutex = PTHREAD_MUTEX_INITIALIZER, .psi_mutex = nullptr }
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 #else  // __FreeBSD__, __linux__, at least
 #define TOKU_MUTEX_ADAPTIVE PTHREAD_MUTEX_ADAPTIVE_NP
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
 #define TOKU_ADAPTIVE_MUTEX_INITIALIZER                                        \
     {                                                                          \
         .pmutex = PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP, .psi_mutex = nullptr, \
@@ -152,8 +152,8 @@ typedef struct toku_mutex_aligned {
 #else
 #define TOKU_ADAPTIVE_MUTEX_INITIALIZER \
     { .pmutex = PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP, .psi_mutex = nullptr }
-#endif
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
+#endif  // defined(__APPLE__)
 
 // Different OSes implement mutexes as different amounts of nested structs.
 // C++ will fill out all missing values with zeroes if you provide at least one
@@ -188,7 +188,7 @@ toku_mutexattr_destroy(toku_pthread_mutexattr_t *attr) {
     assert_zero(r);
 }
 
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
 static inline void toku_mutex_assert_locked(const toku_mutex_t *mutex) {
     invariant(mutex->locked);
     invariant(mutex->owner == pthread_self());
@@ -197,7 +197,7 @@ static inline void toku_mutex_assert_locked(const toku_mutex_t *mutex) {
 static inline void
 toku_mutex_assert_locked(const toku_mutex_t *mutex __attribute__((unused))) {
 }
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 
 // asserting that a mutex is unlocked only makes sense
 // if the calling thread can guaruntee that no other threads
@@ -207,7 +207,7 @@ toku_mutex_assert_locked(const toku_mutex_t *mutex __attribute__((unused))) {
 // when a node is locked the caller knows that no other threads
 // can be trying to lock its childrens' mutexes. the children
 // are in one of two fixed states: locked or unlocked.
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
 static inline void
 toku_mutex_assert_unlocked(toku_mutex_t *mutex) {
     invariant(mutex->owner == 0);
@@ -216,7 +216,7 @@ toku_mutex_assert_unlocked(toku_mutex_t *mutex) {
 #else
 static inline void toku_mutex_assert_unlocked(toku_mutex_t *mutex
                                               __attribute__((unused))) {}
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 
 #define toku_mutex_lock(M) \
     toku_mutex_lock_with_source_location(M, __FILE__, __LINE__)
@@ -231,13 +231,13 @@ static inline void toku_cond_init(toku_cond_t *cond,
     toku_mutex_trylock_with_source_location(M, __FILE__, __LINE__)
 
 inline void toku_mutex_unlock(toku_mutex_t *mutex) {
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(mutex->owner == pthread_self());
     invariant(mutex->valid);
     invariant(mutex->locked);
     mutex->locked = false;
     mutex->owner = 0;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
     toku_instr_mutex_unlock(mutex->psi_mutex);
     int r = pthread_mutex_unlock(&mutex->pmutex);
     assert_zero(r);
@@ -254,13 +254,13 @@ inline void toku_mutex_lock_with_source_location(toku_mutex_t *mutex,
     toku_instr_mutex_lock_end(mutex_instr, r);
 
     assert_zero(r);
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(mutex->valid);
     invariant(!mutex->locked);
     invariant(mutex->owner == 0);
     mutex->locked = true;
     mutex->owner = pthread_self();
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 }
 
 inline int toku_mutex_trylock_with_source_location(toku_mutex_t *mutex,
@@ -273,7 +273,7 @@ inline int toku_mutex_trylock_with_source_location(toku_mutex_t *mutex,
     const int r = pthread_mutex_lock(&mutex->pmutex);
     toku_instr_mutex_lock_end(mutex_instr, r);
 
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     if (r == 0) {
         invariant(mutex->valid);
         invariant(!mutex->locked);
@@ -281,7 +281,7 @@ inline int toku_mutex_trylock_with_source_location(toku_mutex_t *mutex,
         mutex->locked = true;
         mutex->owner = pthread_self();
     }
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
     return r;
 }
 
@@ -310,11 +310,11 @@ inline void toku_cond_wait_with_source_location(toku_cond_t *cond,
                                                 const char *src_file,
                                                 uint src_line) {
 
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(mutex->locked);
     mutex->locked = false;
     mutex->owner = 0;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 
     /* Instrumentation start */
     toku_cond_instrumentation cond_instr;
@@ -332,11 +332,11 @@ inline void toku_cond_wait_with_source_location(toku_cond_t *cond,
     toku_instr_cond_wait_end(cond_instr, r);
 
     assert_zero(r);
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(!mutex->locked);
     mutex->locked = true;
     mutex->owner = pthread_self();
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 }
 
 inline int toku_cond_timedwait_with_source_location(toku_cond_t *cond,
@@ -344,11 +344,11 @@ inline int toku_cond_timedwait_with_source_location(toku_cond_t *cond,
                                                     toku_timespec_t *wakeup_at,
                                                     const char *src_file,
                                                     uint src_line) {
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(mutex->locked);
     mutex->locked = false;
     mutex->owner = 0;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 
     /* Instrumentation start */
     toku_cond_instrumentation cond_instr;
@@ -366,11 +366,11 @@ inline int toku_cond_timedwait_with_source_location(toku_cond_t *cond,
     /* Instrumentation end */
     toku_instr_cond_wait_end(cond_instr, r);
 
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(!mutex->locked);
     mutex->locked = true;
     mutex->owner = pthread_self();
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
     return r;
 }
 
@@ -389,26 +389,26 @@ inline void toku_cond_broadcast(toku_cond_t *cond) {
 inline void toku_mutex_init(const toku_instr_key &key,
                             toku_mutex_t *mutex,
                             const toku_pthread_mutexattr_t *attr) {
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     mutex->valid = true;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
     toku_instr_mutex_init(key, *mutex);
     const int r = pthread_mutex_init(&mutex->pmutex, attr);
     assert_zero(r);
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     mutex->locked = false;
     invariant(mutex->valid);
     mutex->valid = true;
     mutex->owner = 0;
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
 }
 
 inline void toku_mutex_destroy(toku_mutex_t *mutex) {
-#if TOKU_PTHREAD_DEBUG
+#if defined(TOKU_PTHREAD_DEBUG)
     invariant(mutex->valid);
     mutex->valid = false;
     invariant(!mutex->locked);
-#endif
+#endif  // defined(TOKU_PTHREAD_DEBUG)
     toku_instr_mutex_destroy(mutex->psi_mutex);
     int r = pthread_mutex_destroy(&mutex->pmutex);
     assert_zero(r);
diff --git a/storage/tokudb/PerconaFT/tools/CMakeLists.txt b/storage/tokudb/PerconaFT/tools/CMakeLists.txt
index d54c2c21827..710a55a5957 100644
--- a/storage/tokudb/PerconaFT/tools/CMakeLists.txt
+++ b/storage/tokudb/PerconaFT/tools/CMakeLists.txt
@@ -15,16 +15,12 @@ foreach(tool ${tools})
     if ((CMAKE_BUILD_TYPE MATCHES "Debug") AND
         (CMAKE_CXX_FLAGS_DEBUG MATCHES " -DENABLED_DEBUG_SYNC"))
       if (MYSQL_BASE_VERSION VERSION_EQUAL "8.0")
-        target_link_libraries(${tool} sql_main sql_gis sql_main binlog rpl master slave ${ICU_LIBRARIES})
+        target_link_libraries(${tool} sql_main sql_gis sql_main sql_dd sql_gis binlog rpl master slave ${ICU_LIBRARIES})
       else ()
         target_link_libraries(${tool} sql binlog rpl master slave)
       endif ()
     else ()
-      if (MYSQL_BASE_VERSION VERSION_EQUAL "8.0")
-        target_link_libraries(${tool} mysqlclient)
-      else ()
-        target_link_libraries(${tool} perconaserverclient)
-      endif ()
+      target_link_libraries(${tool} perconaserverclient)
     endif ()
   endif ()
 
diff --git a/storage/tokudb/PerconaFT/util/growable_array.h b/storage/tokudb/PerconaFT/util/growable_array.h
index e8873ae4abd..ad60ea6395b 100644
--- a/storage/tokudb/PerconaFT/util/growable_array.h
+++ b/storage/tokudb/PerconaFT/util/growable_array.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/PerconaFT/util/omt.cc b/storage/tokudb/PerconaFT/util/omt.cc
index 1fae0712c77..846c4df7f54 100644
--- a/storage/tokudb/PerconaFT/util/omt.cc
+++ b/storage/tokudb/PerconaFT/util/omt.cc
@@ -32,1105 +32,1356 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
-#ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
+#ident \
+    "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
 
-#include <string.h>
 #include <db.h>
+#include <string.h>
 
 #include <portability/memory.h>
 
 namespace toku {
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::create(void) {
-    this->create_internal(2);
-    if (supports_marks) {
-        this->convert_to_tree();
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::create(void) {
+        this->create_internal(2);
+        if (supports_marks) {
+            this->convert_to_tree();
+        }
     }
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::create_no_array(void) {
-    if (!supports_marks) {
-        this->create_internal_no_array(0);
-    } else {
-        this->is_array = false;
-        this->capacity = 0;
-        this->d.t.nodes = nullptr;
-        this->d.t.root.set_to_null();
-        this->d.t.free_idx = 0;
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::create_from_sorted_array(const omtdata_t *const values, const uint32_t numvalues) {
-    this->create_internal(numvalues);
-    memcpy(this->d.a.values, values, numvalues * (sizeof values[0]));
-    this->d.a.num_values = numvalues;
-    if (supports_marks) {
-        this->convert_to_tree();
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::create_steal_sorted_array(omtdata_t **const values, const uint32_t numvalues, const uint32_t new_capacity) {
-    paranoid_invariant_notnull(values);
-    this->create_internal_no_array(new_capacity);
-    this->d.a.num_values = numvalues;
-    this->d.a.values = *values;
-    *values = nullptr;
-    if (supports_marks) {
-        this->convert_to_tree();
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-int omt<omtdata_t, omtdataout_t, supports_marks>::split_at(omt *const newomt, const uint32_t idx) {
-    barf_if_marked(*this);
-    paranoid_invariant_notnull(newomt);
-    if (idx > this->size()) { return EINVAL; }
-    this->convert_to_array();
-    const uint32_t newsize = this->size() - idx;
-    newomt->create_from_sorted_array(&this->d.a.values[this->d.a.start_idx + idx], newsize);
-    this->d.a.num_values = idx;
-    this->maybe_resize_array(idx);
-    if (supports_marks) {
-        this->convert_to_tree();
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::merge(omt *const leftomt, omt *const rightomt) {
-    barf_if_marked(*this);
-    paranoid_invariant_notnull(leftomt);
-    paranoid_invariant_notnull(rightomt);
-    const uint32_t leftsize = leftomt->size();
-    const uint32_t rightsize = rightomt->size();
-    const uint32_t newsize = leftsize + rightsize;
-
-    if (leftomt->is_array) {
-        if (leftomt->capacity - (leftomt->d.a.start_idx + leftomt->d.a.num_values) >= rightsize) {
-            this->create_steal_sorted_array(&leftomt->d.a.values, leftomt->d.a.num_values, leftomt->capacity);
-            this->d.a.start_idx = leftomt->d.a.start_idx;
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::create_no_array(void) {
+        if (!supports_marks) {
+            this->create_internal_no_array(0);
+        } else {
+            this->is_array = false;
+            this->capacity = 0;
+            this->d.t.nodes = nullptr;
+            this->d.t.root.set_to_null();
+            this->d.t.free_idx = 0;
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::create_from_sorted_array(
+        const omtdata_t *const values,
+        const uint32_t numvalues) {
+        this->create_internal(numvalues);
+        memcpy(this->d.a.values, values, numvalues * (sizeof values[0]));
+        this->d.a.num_values = numvalues;
+        if (supports_marks) {
+            this->convert_to_tree();
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void
+    omt<omtdata_t, omtdataout_t, supports_marks>::create_steal_sorted_array(
+        omtdata_t **const values,
+        const uint32_t numvalues,
+        const uint32_t new_capacity) {
+        paranoid_invariant_notnull(values);
+        this->create_internal_no_array(new_capacity);
+        this->d.a.num_values = numvalues;
+        this->d.a.values = *values;
+        *values = nullptr;
+        if (supports_marks) {
+            this->convert_to_tree();
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::split_at(
+        omt *const newomt,
+        const uint32_t idx) {
+        barf_if_marked(*this);
+        paranoid_invariant_notnull(newomt);
+        if (idx > this->size()) {
+            return EINVAL;
+        }
+        this->convert_to_array();
+        const uint32_t newsize = this->size() - idx;
+        newomt->create_from_sorted_array(
+            &this->d.a.values[this->d.a.start_idx + idx], newsize);
+        this->d.a.num_values = idx;
+        this->maybe_resize_array(idx);
+        if (supports_marks) {
+            this->convert_to_tree();
+        }
+        return 0;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::merge(
+        omt *const leftomt,
+        omt *const rightomt) {
+        barf_if_marked(*this);
+        paranoid_invariant_notnull(leftomt);
+        paranoid_invariant_notnull(rightomt);
+        const uint32_t leftsize = leftomt->size();
+        const uint32_t rightsize = rightomt->size();
+        const uint32_t newsize = leftsize + rightsize;
+
+        if (leftomt->is_array) {
+            if (leftomt->capacity -
+                    (leftomt->d.a.start_idx + leftomt->d.a.num_values) >=
+                rightsize) {
+                this->create_steal_sorted_array(&leftomt->d.a.values,
+                                                leftomt->d.a.num_values,
+                                                leftomt->capacity);
+                this->d.a.start_idx = leftomt->d.a.start_idx;
+            } else {
+                this->create_internal(newsize);
+                memcpy(&this->d.a.values[0],
+                       &leftomt->d.a.values[leftomt->d.a.start_idx],
+                       leftomt->d.a.num_values * (sizeof this->d.a.values[0]));
+            }
         } else {
             this->create_internal(newsize);
+            leftomt->fill_array_with_subtree_values(&this->d.a.values[0],
+                                                    leftomt->d.t.root);
+        }
+        leftomt->destroy();
+        this->d.a.num_values = leftsize;
+
+        if (rightomt->is_array) {
+            memcpy(
+                &this->d.a.values[this->d.a.start_idx + this->d.a.num_values],
+                &rightomt->d.a.values[rightomt->d.a.start_idx],
+                rightomt->d.a.num_values * (sizeof this->d.a.values[0]));
+        } else {
+            rightomt->fill_array_with_subtree_values(
+                &this->d.a.values[this->d.a.start_idx + this->d.a.num_values],
+                rightomt->d.t.root);
+        }
+        rightomt->destroy();
+        this->d.a.num_values += rightsize;
+        paranoid_invariant(this->size() == newsize);
+        if (supports_marks) {
+            this->convert_to_tree();
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::clone(const omt &src) {
+        barf_if_marked(*this);
+        this->create_internal(src.size());
+        if (src.is_array) {
             memcpy(&this->d.a.values[0],
-                   &leftomt->d.a.values[leftomt->d.a.start_idx],
-                   leftomt->d.a.num_values * (sizeof this->d.a.values[0]));
-        }
-    } else {
-        this->create_internal(newsize);
-        leftomt->fill_array_with_subtree_values(&this->d.a.values[0], leftomt->d.t.root);
-    }
-    leftomt->destroy();
-    this->d.a.num_values = leftsize;
-
-    if (rightomt->is_array) {
-        memcpy(&this->d.a.values[this->d.a.start_idx + this->d.a.num_values],
-               &rightomt->d.a.values[rightomt->d.a.start_idx],
-               rightomt->d.a.num_values * (sizeof this->d.a.values[0]));
-    } else {
-        rightomt->fill_array_with_subtree_values(&this->d.a.values[this->d.a.start_idx + this->d.a.num_values],
-                                                 rightomt->d.t.root);
-    }
-    rightomt->destroy();
-    this->d.a.num_values += rightsize;
-    paranoid_invariant(this->size() == newsize);
-    if (supports_marks) {
-        this->convert_to_tree();
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::clone(const omt &src) {
-    barf_if_marked(*this);
-    this->create_internal(src.size());
-    if (src.is_array) {
-        memcpy(&this->d.a.values[0], &src.d.a.values[src.d.a.start_idx], src.d.a.num_values * (sizeof this->d.a.values[0]));
-    } else {
-        src.fill_array_with_subtree_values(&this->d.a.values[0], src.d.t.root);
-    }
-    this->d.a.num_values = src.size();
-    if (supports_marks) {
-        this->convert_to_tree();
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::clear(void) {
-    if (this->is_array) {
-        this->d.a.start_idx = 0;
-        this->d.a.num_values = 0;
-    } else {
-        this->d.t.root.set_to_null();
-        this->d.t.free_idx = 0;
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::destroy(void) {
-    this->clear();
-    this->capacity = 0;
-    if (this->is_array) {
-        if (this->d.a.values != nullptr) {
-            toku_free(this->d.a.values);
+                   &src.d.a.values[src.d.a.start_idx],
+                   src.d.a.num_values * (sizeof this->d.a.values[0]));
+        } else {
+            src.fill_array_with_subtree_values(&this->d.a.values[0],
+                                               src.d.t.root);
         }
-        this->d.a.values = nullptr;
-    } else {
-        if (this->d.t.nodes != nullptr) {
-            toku_free(this->d.t.nodes);
+        this->d.a.num_values = src.size();
+        if (supports_marks) {
+            this->convert_to_tree();
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::clear(void) {
+        if (this->is_array) {
+            this->d.a.start_idx = 0;
+            this->d.a.num_values = 0;
+        } else {
+            this->d.t.root.set_to_null();
+            this->d.t.free_idx = 0;
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::destroy(void) {
+        this->clear();
+        this->capacity = 0;
+        if (this->is_array) {
+            if (this->d.a.values != nullptr) {
+                toku_free(this->d.a.values);
+            }
+            this->d.a.values = nullptr;
+        } else {
+            if (this->d.t.nodes != nullptr) {
+                toku_free(this->d.t.nodes);
+            }
+            this->d.t.nodes = nullptr;
         }
-        this->d.t.nodes = nullptr;
     }
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-uint32_t omt<omtdata_t, omtdataout_t, supports_marks>::size(void) const {
-    if (this->is_array) {
-        return this->d.a.num_values;
-    } else {
-        return this->nweight(this->d.t.root);
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    uint32_t omt<omtdata_t, omtdataout_t, supports_marks>::size(void) const {
+        if (this->is_array) {
+            return this->d.a.num_values;
+        } else {
+            return this->nweight(this->d.t.root);
+        }
     }
-}
 
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::insert(
+        const omtdata_t &value,
+        const omtcmp_t &v,
+        uint32_t *const idx) {
+        int r;
+        uint32_t insert_idx;
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::insert(const omtdata_t &value, const omtcmp_t &v, uint32_t *const idx) {
-    int r;
-    uint32_t insert_idx;
+        r = this->find_zero<omtcmp_t, h>(v, nullptr, &insert_idx);
+        if (r == 0) {
+            if (idx)
+                *idx = insert_idx;
+            return DB_KEYEXIST;
+        }
+        if (r != DB_NOTFOUND)
+            return r;
 
-    r = this->find_zero<omtcmp_t, h>(v, nullptr, &insert_idx);
-    if (r==0) {
-        if (idx) *idx = insert_idx;
-        return DB_KEYEXIST;
+        if ((r = this->insert_at(value, insert_idx)))
+            return r;
+        if (idx)
+            *idx = insert_idx;
+
+        return 0;
     }
-    if (r != DB_NOTFOUND) return r;
 
-    if ((r = this->insert_at(value, insert_idx))) return r;
-    if (idx) *idx = insert_idx;
+    // The following 3 functions implement a static if for us.
+    template <typename omtdata_t, typename omtdataout_t>
+    static void barf_if_marked(
+        const omt<omtdata_t, omtdataout_t, false> &UU(omt)) {}
 
-    return 0;
-}
+    template <typename omtdata_t, typename omtdataout_t>
+    static void barf_if_marked(const omt<omtdata_t, omtdataout_t, true> &omt) {
+        invariant(!omt.has_marks());
+    }
 
-// The following 3 functions implement a static if for us.
-template<typename omtdata_t, typename omtdataout_t>
-static void barf_if_marked(const omt<omtdata_t, omtdataout_t, false> &UU(omt)) {
-}
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    bool omt<omtdata_t, omtdataout_t, supports_marks>::has_marks(void) const {
+        static_assert(supports_marks, "Does not support marks");
+        if (this->d.t.root.is_null()) {
+            return false;
+        }
+        const omt_node &node = this->d.t.nodes[this->d.t.root.get_index()];
+        return node.get_marks_below() || node.get_marked();
+    }
 
-template<typename omtdata_t, typename omtdataout_t>
-static void barf_if_marked(const omt<omtdata_t, omtdataout_t, true> &omt) {
-    invariant(!omt.has_marks());
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-bool omt<omtdata_t, omtdataout_t, supports_marks>::has_marks(void) const {
-    static_assert(supports_marks, "Does not support marks");
-    if (this->d.t.root.is_null()) {
-        return false;
-    }
-    const omt_node &node = this->d.t.nodes[this->d.t.root.get_index()];
-    return node.get_marks_below() || node.get_marked();
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-int omt<omtdata_t, omtdataout_t, supports_marks>::insert_at(const omtdata_t &value, const uint32_t idx) {
-    barf_if_marked(*this);
-    if (idx > this->size()) { return EINVAL; }
-
-    this->maybe_resize_or_convert(this->size() + 1);
-    if (this->is_array && idx != this->d.a.num_values &&
-        (idx != 0 || this->d.a.start_idx == 0)) {
-        this->convert_to_tree();
-    }
-    if (this->is_array) {
-        if (idx == this->d.a.num_values) {
-            this->d.a.values[this->d.a.start_idx + this->d.a.num_values] = value;
-        }
-        else {
-            this->d.a.values[--this->d.a.start_idx] = value;
-        }
-        this->d.a.num_values++;
-    }
-    else {
-        subtree *rebalance_subtree = nullptr;
-        this->insert_internal(&this->d.t.root, value, idx, &rebalance_subtree);
-        if (rebalance_subtree != nullptr) {
-            this->rebalance(rebalance_subtree);
-        }
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-int omt<omtdata_t, omtdataout_t, supports_marks>::set_at(const omtdata_t &value, const uint32_t idx) {
-    barf_if_marked(*this);
-    if (idx >= this->size()) { return EINVAL; }
-
-    if (this->is_array) {
-        this->set_at_internal_array(value, idx);
-    } else {
-        this->set_at_internal(this->d.t.root, value, idx);
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-int omt<omtdata_t, omtdataout_t, supports_marks>::delete_at(const uint32_t idx) {
-    barf_if_marked(*this);
-    if (idx >= this->size()) { return EINVAL; }
-
-    this->maybe_resize_or_convert(this->size() - 1);
-    if (this->is_array && idx != 0 && idx != this->d.a.num_values - 1) {
-        this->convert_to_tree();
-    }
-    if (this->is_array) {
-        //Testing for 0 does not rule out it being the last entry.
-        //Test explicitly for num_values-1
-        if (idx != this->d.a.num_values - 1) {
-            this->d.a.start_idx++;
-        }
-        this->d.a.num_values--;
-    } else {
-        subtree *rebalance_subtree = nullptr;
-        this->delete_internal(&this->d.t.root, idx, nullptr, &rebalance_subtree);
-        if (rebalance_subtree != nullptr) {
-            this->rebalance(rebalance_subtree);
-        }
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate(iterate_extra_t *const iterate_extra) const {
-    return this->iterate_on_range<iterate_extra_t, f>(0, this->size(), iterate_extra);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_on_range(const uint32_t left, const uint32_t right, iterate_extra_t *const iterate_extra) const {
-    if (right > this->size()) { return EINVAL; }
-    if (left == right) { return 0; }
-    if (this->is_array) {
-        return this->iterate_internal_array<iterate_extra_t, f>(left, right, iterate_extra);
-    }
-    return this->iterate_internal<iterate_extra_t, f>(left, right, this->d.t.root, 0, iterate_extra);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_and_mark_range(const uint32_t left, const uint32_t right, iterate_extra_t *const iterate_extra) {
-    static_assert(supports_marks, "does not support marks");
-    if (right > this->size()) { return EINVAL; }
-    if (left == right) { return 0; }
-    paranoid_invariant(!this->is_array);
-    return this->iterate_and_mark_range_internal<iterate_extra_t, f>(left, right, this->d.t.root, 0, iterate_extra);
-}
-
-//TODO: We can optimize this if we steal 3 bits.  1 bit: this node is marked.  1 bit: left subtree has marks. 1 bit: right subtree has marks.
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_over_marked(iterate_extra_t *const iterate_extra) const {
-    static_assert(supports_marks, "does not support marks");
-    paranoid_invariant(!this->is_array);
-    return this->iterate_over_marked_internal<iterate_extra_t, f>(this->d.t.root, 0, iterate_extra);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::unmark(const subtree &subtree, const uint32_t index, GrowableArray<node_idx> *const indexes) {
-    if (subtree.is_null()) { return; }
-    omt_node &n = this->d.t.nodes[subtree.get_index()];
-    const uint32_t index_root = index + this->nweight(n.left);
-
-    const bool below = n.get_marks_below();
-    if (below) {
-        this->unmark(n.left, index, indexes);
-    }
-    if (n.get_marked()) {
-        indexes->push(index_root);
-    }
-    n.clear_stolen_bits();
-    if (below) {
-        this->unmark(n.right, index_root + 1, indexes);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::delete_all_marked(void) {
-    static_assert(supports_marks, "does not support marks");
-    if (!this->has_marks()) {
-        return;
-    }
-    paranoid_invariant(!this->is_array);
-    GrowableArray<node_idx> marked_indexes;
-    marked_indexes.init();
-
-    // Remove all marks.
-    // We need to delete all the stolen bits before calling delete_at to prevent barfing.
-    this->unmark(this->d.t.root, 0, &marked_indexes);
-
-    for (uint32_t i = 0; i < marked_indexes.get_size(); i++) {
-        // Delete from left to right, shift by number already deleted.
-        // Alternative is delete from right to left.
-        int r = this->delete_at(marked_indexes.fetch_unchecked(i) - i);
-        lazy_assert_zero(r);
-    }
-    marked_indexes.deinit();
-    barf_if_marked(*this);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-uint32_t omt<omtdata_t, omtdataout_t, supports_marks>::verify_marks_consistent_internal(const subtree &subtree, const bool UU(allow_marks)) const {
-    if (subtree.is_null()) {
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::insert_at(
+        const omtdata_t &value,
+        const uint32_t idx) {
+        barf_if_marked(*this);
+        if (idx > this->size()) {
+            return EINVAL;
+        }
+
+        this->maybe_resize_or_convert(this->size() + 1);
+        if (this->is_array && idx != this->d.a.num_values &&
+            (idx != 0 || this->d.a.start_idx == 0)) {
+            this->convert_to_tree();
+        }
+        if (this->is_array) {
+            if (idx == this->d.a.num_values) {
+                this->d.a.values[this->d.a.start_idx + this->d.a.num_values] =
+                    value;
+            } else {
+                this->d.a.values[--this->d.a.start_idx] = value;
+            }
+            this->d.a.num_values++;
+        } else {
+            subtree *rebalance_subtree = nullptr;
+            this->insert_internal(
+                &this->d.t.root, value, idx, &rebalance_subtree);
+            if (rebalance_subtree != nullptr) {
+                this->rebalance(rebalance_subtree);
+            }
+        }
         return 0;
     }
-    const omt_node &node = this->d.t.nodes[subtree.get_index()];
-    uint32_t num_marks = verify_marks_consistent_internal(node.left, node.get_marks_below());
-    num_marks += verify_marks_consistent_internal(node.right, node.get_marks_below());
-    if (node.get_marks_below()) {
-        paranoid_invariant(allow_marks);
-        paranoid_invariant(num_marks > 0);
-    } else {
-        // redundant with invariant below, but nice to have explicitly
-        paranoid_invariant(num_marks == 0);
-    }
-    if (node.get_marked()) {
-        paranoid_invariant(allow_marks);
-        ++num_marks;
-    }
-    return num_marks;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::verify_marks_consistent(void) const {
-    static_assert(supports_marks, "does not support marks");
-    paranoid_invariant(!this->is_array);
-    this->verify_marks_consistent_internal(this->d.t.root, true);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(omtdata_t *, const uint32_t, iterate_extra_t *const)>
-void omt<omtdata_t, omtdataout_t, supports_marks>::iterate_ptr(iterate_extra_t *const iterate_extra) {
-    if (this->is_array) {
-        this->iterate_ptr_internal_array<iterate_extra_t, f>(0, this->size(), iterate_extra);
-    } else {
-        this->iterate_ptr_internal<iterate_extra_t, f>(0, this->size(), this->d.t.root, 0, iterate_extra);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-int omt<omtdata_t, omtdataout_t, supports_marks>::fetch(const uint32_t idx, omtdataout_t *const value) const {
-    if (idx >= this->size()) { return EINVAL; }
-    if (this->is_array) {
-        this->fetch_internal_array(idx, value);
-    } else {
-        this->fetch_internal(this->d.t.root, idx, value);
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_zero(const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    uint32_t tmp_index;
-    uint32_t *const child_idxp = (idxp != nullptr) ? idxp : &tmp_index;
-    int r;
-    if (this->is_array) {
-        r = this->find_internal_zero_array<omtcmp_t, h>(extra, value, child_idxp);
-    }
-    else {
-        r = this->find_internal_zero<omtcmp_t, h>(this->d.t.root, extra, value, child_idxp);
-    }
-    return r;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find(const omtcmp_t &extra, int direction, omtdataout_t *const value, uint32_t *const idxp) const {
-    uint32_t tmp_index;
-    uint32_t *const child_idxp = (idxp != nullptr) ? idxp : &tmp_index;
-    paranoid_invariant(direction != 0);
-    if (direction < 0) {
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::set_at(
+        const omtdata_t &value,
+        const uint32_t idx) {
+        barf_if_marked(*this);
+        if (idx >= this->size()) {
+            return EINVAL;
+        }
+
         if (this->is_array) {
-            return this->find_internal_minus_array<omtcmp_t, h>(extra, value, child_idxp);
+            this->set_at_internal_array(value, idx);
         } else {
-            return this->find_internal_minus<omtcmp_t, h>(this->d.t.root, extra, value, child_idxp);
+            this->set_at_internal(this->d.t.root, value, idx);
+        }
+        return 0;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::delete_at(
+        const uint32_t idx) {
+        barf_if_marked(*this);
+        if (idx >= this->size()) {
+            return EINVAL;
+        }
+
+        this->maybe_resize_or_convert(this->size() - 1);
+        if (this->is_array && idx != 0 && idx != this->d.a.num_values - 1) {
+            this->convert_to_tree();
         }
-    } else {
         if (this->is_array) {
-            return this->find_internal_plus_array<omtcmp_t, h>(extra, value, child_idxp);
+            // Testing for 0 does not rule out it being the last entry.
+            // Test explicitly for num_values-1
+            if (idx != this->d.a.num_values - 1) {
+                this->d.a.start_idx++;
+            }
+            this->d.a.num_values--;
         } else {
-            return this->find_internal_plus<omtcmp_t, h>(this->d.t.root, extra, value, child_idxp);
+            subtree *rebalance_subtree = nullptr;
+            this->delete_internal(
+                &this->d.t.root, idx, nullptr, &rebalance_subtree);
+            if (rebalance_subtree != nullptr) {
+                this->rebalance(rebalance_subtree);
+            }
         }
+        return 0;
     }
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-size_t omt<omtdata_t, omtdataout_t, supports_marks>::memory_size(void) {
-    if (this->is_array) {
-        return (sizeof *this) + this->capacity * (sizeof this->d.a.values[0]);
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::iterate(
+        iterate_extra_t *const iterate_extra) const {
+        return this->iterate_on_range<iterate_extra_t, f>(
+            0, this->size(), iterate_extra);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_on_range(
+        const uint32_t left,
+        const uint32_t right,
+        iterate_extra_t *const iterate_extra) const {
+        if (right > this->size()) {
+            return EINVAL;
+        }
+        if (left == right) {
+            return 0;
+        }
+        if (this->is_array) {
+            return this->iterate_internal_array<iterate_extra_t, f>(
+                left, right, iterate_extra);
+        }
+        return this->iterate_internal<iterate_extra_t, f>(
+            left, right, this->d.t.root, 0, iterate_extra);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_and_mark_range(
+        const uint32_t left,
+        const uint32_t right,
+        iterate_extra_t *const iterate_extra) {
+        static_assert(supports_marks, "does not support marks");
+        if (right > this->size()) {
+            return EINVAL;
+        }
+        if (left == right) {
+            return 0;
+        }
+        paranoid_invariant(!this->is_array);
+        return this->iterate_and_mark_range_internal<iterate_extra_t, f>(
+            left, right, this->d.t.root, 0, iterate_extra);
+    }
+
+    // TODO: We can optimize this if we steal 3 bits.  1 bit: this node is
+    // marked.  1 bit: left subtree has marks. 1 bit: right subtree has marks.
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_over_marked(
+        iterate_extra_t *const iterate_extra) const {
+        static_assert(supports_marks, "does not support marks");
+        paranoid_invariant(!this->is_array);
+        return this->iterate_over_marked_internal<iterate_extra_t, f>(
+            this->d.t.root, 0, iterate_extra);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::unmark(
+        const subtree &st,
+        const uint32_t index,
+        GrowableArray<node_idx> *const indexes) {
+        if (st.is_null()) {
+            return;
+        }
+        omt_node &n = this->d.t.nodes[st.get_index()];
+        const uint32_t index_root = index + this->nweight(n.left);
+
+        const bool below = n.get_marks_below();
+        if (below) {
+            this->unmark(n.left, index, indexes);
+        }
+        if (n.get_marked()) {
+            indexes->push(index_root);
+        }
+        n.clear_stolen_bits();
+        if (below) {
+            this->unmark(n.right, index_root + 1, indexes);
+        }
     }
-    return (sizeof *this) + this->capacity * (sizeof this->d.t.nodes[0]);
-}
 
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::delete_all_marked(void) {
+        static_assert(supports_marks, "does not support marks");
+        if (!this->has_marks()) {
+            return;
+        }
+        paranoid_invariant(!this->is_array);
+        GrowableArray<node_idx> marked_indexes;
+        marked_indexes.init();
+
+        // Remove all marks.
+        // We need to delete all the stolen bits before calling delete_at to
+        // prevent barfing.
+        this->unmark(this->d.t.root, 0, &marked_indexes);
+
+        for (uint32_t i = 0; i < marked_indexes.get_size(); i++) {
+            // Delete from left to right, shift by number already deleted.
+            // Alternative is delete from right to left.
+            int r = this->delete_at(marked_indexes.fetch_unchecked(i) - i);
+            lazy_assert_zero(r);
+        }
+        marked_indexes.deinit();
+        barf_if_marked(*this);
+    }
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::create_internal_no_array(const uint32_t new_capacity) {
-    this->is_array = true;
-    this->d.a.start_idx = 0;
-    this->d.a.num_values = 0;
-    this->d.a.values = nullptr;
-    this->capacity = new_capacity;
-}
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    uint32_t omt<omtdata_t, omtdataout_t, supports_marks>::
+        verify_marks_consistent_internal(const subtree &st,
+                                         const bool UU(allow_marks)) const {
+        if (st.is_null()) {
+            return 0;
+        }
+        const omt_node &node = this->d.t.nodes[st.get_index()];
+        uint32_t num_marks =
+            verify_marks_consistent_internal(node.left, node.get_marks_below());
+        num_marks += verify_marks_consistent_internal(node.right,
+                                                      node.get_marks_below());
+        if (node.get_marks_below()) {
+            paranoid_invariant(allow_marks);
+            paranoid_invariant(num_marks > 0);
+        } else {
+            // redundant with invariant below, but nice to have explicitly
+            paranoid_invariant(num_marks == 0);
+        }
+        if (node.get_marked()) {
+            paranoid_invariant(allow_marks);
+            ++num_marks;
+        }
+        return num_marks;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::verify_marks_consistent(
+        void) const {
+        static_assert(supports_marks, "does not support marks");
+        paranoid_invariant(!this->is_array);
+        this->verify_marks_consistent_internal(this->d.t.root, true);
+    }
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::create_internal(const uint32_t new_capacity) {
-    this->create_internal_no_array(new_capacity);
-    XMALLOC_N(this->capacity, this->d.a.values);
-}
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename iterate_extra_t,
+              int (*f)(omtdata_t *, const uint32_t, iterate_extra_t *const)>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::iterate_ptr(
+        iterate_extra_t *const iterate_extra) {
+        if (this->is_array) {
+            this->iterate_ptr_internal_array<iterate_extra_t, f>(
+                0, this->size(), iterate_extra);
+        } else {
+            this->iterate_ptr_internal<iterate_extra_t, f>(
+                0, this->size(), this->d.t.root, 0, iterate_extra);
+        }
+    }
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-uint32_t omt<omtdata_t, omtdataout_t, supports_marks>::nweight(const subtree &subtree) const {
-    if (subtree.is_null()) {
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::fetch(
+        const uint32_t idx,
+        omtdataout_t *const value) const {
+        if (idx >= this->size()) {
+            return EINVAL;
+        }
+        if (this->is_array) {
+            this->fetch_internal_array(idx, value);
+        } else {
+            this->fetch_internal(this->d.t.root, idx, value);
+        }
         return 0;
-    } else {
-        return this->d.t.nodes[subtree.get_index()].weight;
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-typename omt<omtdata_t, omtdataout_t, supports_marks>::node_idx omt<omtdata_t, omtdataout_t, supports_marks>::node_malloc(void) {
-    paranoid_invariant(this->d.t.free_idx < this->capacity);
-    omt_node &n = this->d.t.nodes[this->d.t.free_idx];
-    n.clear_stolen_bits();
-    return this->d.t.free_idx++;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::node_free(const node_idx UU(idx)) {
-    paranoid_invariant(idx < this->capacity);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::maybe_resize_array(const uint32_t n) {
-    const uint32_t new_size = n<=2 ? 4 : 2*n;
-    const uint32_t room = this->capacity - this->d.a.start_idx;
-
-    if (room < n || this->capacity / 2 >= new_size) {
-        omtdata_t *XMALLOC_N(new_size, tmp_values);
-        memcpy(tmp_values, &this->d.a.values[this->d.a.start_idx],
-               this->d.a.num_values * (sizeof tmp_values[0]));
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_zero(
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        uint32_t tmp_index;
+        uint32_t *const child_idxp = (idxp != nullptr) ? idxp : &tmp_index;
+        int r;
+        if (this->is_array) {
+            r = this->find_internal_zero_array<omtcmp_t, h>(
+                extra, value, child_idxp);
+        } else {
+            r = this->find_internal_zero<omtcmp_t, h>(
+                this->d.t.root, extra, value, child_idxp);
+        }
+        return r;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find(
+        const omtcmp_t &extra,
+        int direction,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        uint32_t tmp_index;
+        uint32_t *const child_idxp = (idxp != nullptr) ? idxp : &tmp_index;
+        paranoid_invariant(direction != 0);
+        if (direction < 0) {
+            if (this->is_array) {
+                return this->find_internal_minus_array<omtcmp_t, h>(
+                    extra, value, child_idxp);
+            } else {
+                return this->find_internal_minus<omtcmp_t, h>(
+                    this->d.t.root, extra, value, child_idxp);
+            }
+        } else {
+            if (this->is_array) {
+                return this->find_internal_plus_array<omtcmp_t, h>(
+                    extra, value, child_idxp);
+            } else {
+                return this->find_internal_plus<omtcmp_t, h>(
+                    this->d.t.root, extra, value, child_idxp);
+            }
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    size_t omt<omtdata_t, omtdataout_t, supports_marks>::memory_size(void) {
+        if (this->is_array) {
+            return (sizeof *this) +
+                   this->capacity * (sizeof this->d.a.values[0]);
+        }
+        return (sizeof *this) + this->capacity * (sizeof this->d.t.nodes[0]);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::create_internal_no_array(
+        const uint32_t new_capacity) {
+        this->is_array = true;
         this->d.a.start_idx = 0;
-        this->capacity = new_size;
-        toku_free(this->d.a.values);
-        this->d.a.values = tmp_values;
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::fill_array_with_subtree_values(omtdata_t *const array, const subtree &subtree) const {
-    if (subtree.is_null()) return;
-    const omt_node &tree = this->d.t.nodes[subtree.get_index()];
-    this->fill_array_with_subtree_values(&array[0], tree.left);
-    array[this->nweight(tree.left)] = tree.value;
-    this->fill_array_with_subtree_values(&array[this->nweight(tree.left) + 1], tree.right);
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::convert_to_array(void) {
-    if (!this->is_array) {
-        const uint32_t num_values = this->size();
-        uint32_t new_size = 2*num_values;
-        new_size = new_size < 4 ? 4 : new_size;
-
-        omtdata_t *XMALLOC_N(new_size, tmp_values);
-        this->fill_array_with_subtree_values(tmp_values, this->d.t.root);
-        toku_free(this->d.t.nodes);
-        this->is_array       = true;
-        this->capacity       = new_size;
-        this->d.a.num_values = num_values;
-        this->d.a.values     = tmp_values;
-        this->d.a.start_idx  = 0;
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::rebuild_from_sorted_array(subtree *const subtree, const omtdata_t *const values, const uint32_t numvalues) {
-    if (numvalues==0) {
-        subtree->set_to_null();
-    } else {
-        const uint32_t halfway = numvalues/2;
-        const node_idx newidx = this->node_malloc();
-        omt_node *const newnode = &this->d.t.nodes[newidx];
-        newnode->weight = numvalues;
-        newnode->value = values[halfway];
-        subtree->set_index(newidx);
-        // update everything before the recursive calls so the second call can be a tail call.
-        this->rebuild_from_sorted_array(&newnode->left,  &values[0], halfway);
-        this->rebuild_from_sorted_array(&newnode->right, &values[halfway+1], numvalues - (halfway+1));
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::convert_to_tree(void) {
-    if (this->is_array) {
-        const uint32_t num_nodes = this->size();
-        uint32_t new_size  = num_nodes*2;
-        new_size = new_size < 4 ? 4 : new_size;
-
-        omt_node *XMALLOC_N(new_size, new_nodes);
-        omtdata_t *const values = this->d.a.values;
-        omtdata_t *const tmp_values = &values[this->d.a.start_idx];
-        this->is_array = false;
-        this->d.t.nodes = new_nodes;
-        this->capacity = new_size;
-        this->d.t.free_idx = 0;
-        this->d.t.root.set_to_null();
-        this->rebuild_from_sorted_array(&this->d.t.root, tmp_values, num_nodes);
-        toku_free(values);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::maybe_resize_or_convert(const uint32_t n) {
-    if (this->is_array) {
-        this->maybe_resize_array(n);
-    } else {
-        const uint32_t new_size = n<=2 ? 4 : 2*n;
-        const uint32_t num_nodes = this->nweight(this->d.t.root);
-        if ((this->capacity/2 >= new_size) ||
-            (this->d.t.free_idx >= this->capacity && num_nodes < n) ||
-            (this->capacity<n)) {
-            this->convert_to_array();
-            // if we had a free list, the "supports_marks" version could
-            // just resize, as it is now, we have to convert to and back
-            // from an array.
-            if (supports_marks) {
-                this->convert_to_tree();
+        this->d.a.num_values = 0;
+        this->d.a.values = nullptr;
+        this->capacity = new_capacity;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::create_internal(
+        const uint32_t new_capacity) {
+        this->create_internal_no_array(new_capacity);
+        XMALLOC_N(this->capacity, this->d.a.values);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    uint32_t omt<omtdata_t, omtdataout_t, supports_marks>::nweight(
+        const subtree &st) const {
+        if (st.is_null()) {
+            return 0;
+        } else {
+            return this->d.t.nodes[st.get_index()].weight;
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    typename omt<omtdata_t, omtdataout_t, supports_marks>::node_idx
+    omt<omtdata_t, omtdataout_t, supports_marks>::node_malloc(void) {
+        paranoid_invariant(this->d.t.free_idx < this->capacity);
+        omt_node &n = this->d.t.nodes[this->d.t.free_idx];
+        n.clear_stolen_bits();
+        return this->d.t.free_idx++;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::node_free(
+        const node_idx UU(idx)) {
+        paranoid_invariant(idx < this->capacity);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::maybe_resize_array(
+        const uint32_t n) {
+        const uint32_t new_size = n <= 2 ? 4 : 2 * n;
+        const uint32_t room = this->capacity - this->d.a.start_idx;
+
+        if (room < n || this->capacity / 2 >= new_size) {
+            omtdata_t *XMALLOC_N(new_size, tmp_values);
+            memcpy(tmp_values,
+                   &this->d.a.values[this->d.a.start_idx],
+                   this->d.a.num_values * (sizeof tmp_values[0]));
+            this->d.a.start_idx = 0;
+            this->capacity = new_size;
+            toku_free(this->d.a.values);
+            this->d.a.values = tmp_values;
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::
+        fill_array_with_subtree_values(omtdata_t *const array,
+                                       const subtree &st) const {
+        if (st.is_null())
+            return;
+        const omt_node &tree = this->d.t.nodes[st.get_index()];
+        this->fill_array_with_subtree_values(&array[0], tree.left);
+        array[this->nweight(tree.left)] = tree.value;
+        this->fill_array_with_subtree_values(
+            &array[this->nweight(tree.left) + 1], tree.right);
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::convert_to_array(void) {
+        if (!this->is_array) {
+            const uint32_t num_values = this->size();
+            uint32_t new_size = 2 * num_values;
+            new_size = new_size < 4 ? 4 : new_size;
+
+            omtdata_t *XMALLOC_N(new_size, tmp_values);
+            this->fill_array_with_subtree_values(tmp_values, this->d.t.root);
+            toku_free(this->d.t.nodes);
+            this->is_array = true;
+            this->capacity = new_size;
+            this->d.a.num_values = num_values;
+            this->d.a.values = tmp_values;
+            this->d.a.start_idx = 0;
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void
+    omt<omtdata_t, omtdataout_t, supports_marks>::rebuild_from_sorted_array(
+        subtree *const st,
+        const omtdata_t *const values,
+        const uint32_t numvalues) {
+        if (numvalues == 0) {
+            st->set_to_null();
+        } else {
+            const uint32_t halfway = numvalues / 2;
+            const node_idx newidx = this->node_malloc();
+            omt_node *const newnode = &this->d.t.nodes[newidx];
+            newnode->weight = numvalues;
+            newnode->value = values[halfway];
+            st->set_index(newidx);
+            // update everything before the recursive calls so the second call
+            // can be a tail call.
+            this->rebuild_from_sorted_array(
+                &newnode->left, &values[0], halfway);
+            this->rebuild_from_sorted_array(&newnode->right,
+                                            &values[halfway + 1],
+                                            numvalues - (halfway + 1));
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::convert_to_tree(void) {
+        if (this->is_array) {
+            const uint32_t num_nodes = this->size();
+            uint32_t new_size = num_nodes * 2;
+            new_size = new_size < 4 ? 4 : new_size;
+
+            omt_node *XMALLOC_N(new_size, new_nodes);
+            omtdata_t *const values = this->d.a.values;
+            omtdata_t *const tmp_values = &values[this->d.a.start_idx];
+            this->is_array = false;
+            this->d.t.nodes = new_nodes;
+            this->capacity = new_size;
+            this->d.t.free_idx = 0;
+            this->d.t.root.set_to_null();
+            this->rebuild_from_sorted_array(
+                &this->d.t.root, tmp_values, num_nodes);
+            toku_free(values);
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::maybe_resize_or_convert(
+        const uint32_t n) {
+        if (this->is_array) {
+            this->maybe_resize_array(n);
+        } else {
+            const uint32_t new_size = n <= 2 ? 4 : 2 * n;
+            const uint32_t num_nodes = this->nweight(this->d.t.root);
+            if ((this->capacity / 2 >= new_size) ||
+                (this->d.t.free_idx >= this->capacity && num_nodes < n) ||
+                (this->capacity < n)) {
+                this->convert_to_array();
+                // if we had a free list, the "supports_marks" version could
+                // just resize, as it is now, we have to convert to and back
+                // from an array.
+                if (supports_marks) {
+                    this->convert_to_tree();
+                }
             }
         }
     }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-bool omt<omtdata_t, omtdataout_t, supports_marks>::will_need_rebalance(const subtree &subtree, const int leftmod, const int rightmod) const {
-    if (subtree.is_null()) { return false; }
-    const omt_node &n = this->d.t.nodes[subtree.get_index()];
-    // one of the 1's is for the root.
-    // the other is to take ceil(n/2)
-    const uint32_t weight_left  = this->nweight(n.left)  + leftmod;
-    const uint32_t weight_right = this->nweight(n.right) + rightmod;
-    return ((1+weight_left < (1+1+weight_right)/2)
-            ||
-            (1+weight_right < (1+1+weight_left)/2));
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::insert_internal(subtree *const subtreep, const omtdata_t &value, const uint32_t idx, subtree **const rebalance_subtree) {
-    if (subtreep->is_null()) {
-        paranoid_invariant_zero(idx);
-        const node_idx newidx = this->node_malloc();
-        omt_node *const newnode = &this->d.t.nodes[newidx];
-        newnode->weight = 1;
-        newnode->left.set_to_null();
-        newnode->right.set_to_null();
-        newnode->value = value;
-        subtreep->set_index(newidx);
-    } else {
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    bool omt<omtdata_t, omtdataout_t, supports_marks>::will_need_rebalance(
+        const subtree &st,
+        const int leftmod,
+        const int rightmod) const {
+        if (st.is_null()) {
+            return false;
+        }
+        const omt_node &n = this->d.t.nodes[st.get_index()];
+        // one of the 1's is for the root.
+        // the other is to take ceil(n/2)
+        const uint32_t weight_left = this->nweight(n.left) + leftmod;
+        const uint32_t weight_right = this->nweight(n.right) + rightmod;
+        return ((1 + weight_left < (1 + 1 + weight_right) / 2) ||
+                (1 + weight_right < (1 + 1 + weight_left) / 2));
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::insert_internal(
+        subtree *const subtreep,
+        const omtdata_t &value,
+        const uint32_t idx,
+        subtree **const rebalance_subtree) {
+        if (subtreep->is_null()) {
+            paranoid_invariant_zero(idx);
+            const node_idx newidx = this->node_malloc();
+            omt_node *const newnode = &this->d.t.nodes[newidx];
+            newnode->weight = 1;
+            newnode->left.set_to_null();
+            newnode->right.set_to_null();
+            newnode->value = value;
+            subtreep->set_index(newidx);
+        } else {
+            omt_node &n = this->d.t.nodes[subtreep->get_index()];
+            n.weight++;
+            if (idx <= this->nweight(n.left)) {
+                if (*rebalance_subtree == nullptr &&
+                    this->will_need_rebalance(*subtreep, 1, 0)) {
+                    *rebalance_subtree = subtreep;
+                }
+                this->insert_internal(&n.left, value, idx, rebalance_subtree);
+            } else {
+                if (*rebalance_subtree == nullptr &&
+                    this->will_need_rebalance(*subtreep, 0, 1)) {
+                    *rebalance_subtree = subtreep;
+                }
+                const uint32_t sub_index = idx - this->nweight(n.left) - 1;
+                this->insert_internal(
+                    &n.right, value, sub_index, rebalance_subtree);
+            }
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::set_at_internal_array(
+        const omtdata_t &value,
+        const uint32_t idx) {
+        this->d.a.values[this->d.a.start_idx + idx] = value;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::set_at_internal(
+        const subtree &st,
+        const omtdata_t &value,
+        const uint32_t idx) {
+        paranoid_invariant(!st.is_null());
+        omt_node &n = this->d.t.nodes[st.get_index()];
+        const uint32_t leftweight = this->nweight(n.left);
+        if (idx < leftweight) {
+            this->set_at_internal(n.left, value, idx);
+        } else if (idx == leftweight) {
+            n.value = value;
+        } else {
+            this->set_at_internal(n.right, value, idx - leftweight - 1);
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::delete_internal(
+        subtree *const subtreep,
+        const uint32_t idx,
+        omt_node *const copyn,
+        subtree **const rebalance_subtree) {
+        paranoid_invariant_notnull(subtreep);
+        paranoid_invariant_notnull(rebalance_subtree);
+        paranoid_invariant(!subtreep->is_null());
         omt_node &n = this->d.t.nodes[subtreep->get_index()];
-        n.weight++;
-        if (idx <= this->nweight(n.left)) {
-            if (*rebalance_subtree == nullptr && this->will_need_rebalance(*subtreep, 1, 0)) {
+        const uint32_t leftweight = this->nweight(n.left);
+        if (idx < leftweight) {
+            n.weight--;
+            if (*rebalance_subtree == nullptr &&
+                this->will_need_rebalance(*subtreep, -1, 0)) {
                 *rebalance_subtree = subtreep;
             }
-            this->insert_internal(&n.left, value, idx, rebalance_subtree);
+            this->delete_internal(&n.left, idx, copyn, rebalance_subtree);
+        } else if (idx == leftweight) {
+            if (n.left.is_null()) {
+                const uint32_t oldidx = subtreep->get_index();
+                *subtreep = n.right;
+                if (copyn != nullptr) {
+                    copyn->value = n.value;
+                }
+                this->node_free(oldidx);
+            } else if (n.right.is_null()) {
+                const uint32_t oldidx = subtreep->get_index();
+                *subtreep = n.left;
+                if (copyn != nullptr) {
+                    copyn->value = n.value;
+                }
+                this->node_free(oldidx);
+            } else {
+                if (*rebalance_subtree == nullptr &&
+                    this->will_need_rebalance(*subtreep, 0, -1)) {
+                    *rebalance_subtree = subtreep;
+                }
+                // don't need to copy up value, it's only used by this
+                // next call, and when that gets to the bottom there
+                // won't be any more recursion
+                n.weight--;
+                this->delete_internal(&n.right, 0, &n, rebalance_subtree);
+            }
         } else {
-            if (*rebalance_subtree == nullptr && this->will_need_rebalance(*subtreep, 0, 1)) {
+            n.weight--;
+            if (*rebalance_subtree == nullptr &&
+                this->will_need_rebalance(*subtreep, 0, -1)) {
                 *rebalance_subtree = subtreep;
             }
-            const uint32_t sub_index = idx - this->nweight(n.left) - 1;
-            this->insert_internal(&n.right, value, sub_index, rebalance_subtree);
-        }
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::set_at_internal_array(const omtdata_t &value, const uint32_t idx) {
-    this->d.a.values[this->d.a.start_idx + idx] = value;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::set_at_internal(const subtree &subtree, const omtdata_t &value, const uint32_t idx) {
-    paranoid_invariant(!subtree.is_null());
-    omt_node &n = this->d.t.nodes[subtree.get_index()];
-    const uint32_t leftweight = this->nweight(n.left);
-    if (idx < leftweight) {
-        this->set_at_internal(n.left, value, idx);
-    } else if (idx == leftweight) {
-        n.value = value;
-    } else {
-        this->set_at_internal(n.right, value, idx - leftweight - 1);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::delete_internal(subtree *const subtreep, const uint32_t idx, omt_node *const copyn, subtree **const rebalance_subtree) {
-    paranoid_invariant_notnull(subtreep);
-    paranoid_invariant_notnull(rebalance_subtree);
-    paranoid_invariant(!subtreep->is_null());
-    omt_node &n = this->d.t.nodes[subtreep->get_index()];
-    const uint32_t leftweight = this->nweight(n.left);
-    if (idx < leftweight) {
-        n.weight--;
-        if (*rebalance_subtree == nullptr && this->will_need_rebalance(*subtreep, -1, 0)) {
-            *rebalance_subtree = subtreep;
-        }
-        this->delete_internal(&n.left, idx, copyn, rebalance_subtree);
-    } else if (idx == leftweight) {
-        if (n.left.is_null()) {
-            const uint32_t oldidx = subtreep->get_index();
-            *subtreep = n.right;
-            if (copyn != nullptr) {
-                copyn->value = n.value;
+            this->delete_internal(
+                &n.right, idx - leftweight - 1, copyn, rebalance_subtree);
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_internal_array(
+        const uint32_t left,
+        const uint32_t right,
+        iterate_extra_t *const iterate_extra) const {
+        int r;
+        for (uint32_t i = left; i < right; ++i) {
+            r = f(this->d.a.values[this->d.a.start_idx + i], i, iterate_extra);
+            if (r != 0) {
+                return r;
             }
-            this->node_free(oldidx);
-        } else if (n.right.is_null()) {
-            const uint32_t oldidx = subtreep->get_index();
-            *subtreep = n.left;
-            if (copyn != nullptr) {
-                copyn->value = n.value;
+        }
+        return 0;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename iterate_extra_t,
+              int (*f)(omtdata_t *, const uint32_t, iterate_extra_t *const)>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::iterate_ptr_internal(
+        const uint32_t left,
+        const uint32_t right,
+        const subtree &st,
+        const uint32_t idx,
+        iterate_extra_t *const iterate_extra) {
+        if (!st.is_null()) {
+            omt_node &n = this->d.t.nodes[st.get_index()];
+            const uint32_t idx_root = idx + this->nweight(n.left);
+            if (left < idx_root) {
+                this->iterate_ptr_internal<iterate_extra_t, f>(
+                    left, right, n.left, idx, iterate_extra);
             }
-            this->node_free(oldidx);
-        } else {
-            if (*rebalance_subtree == nullptr && this->will_need_rebalance(*subtreep, 0, -1)) {
-                *rebalance_subtree = subtreep;
+            if (left <= idx_root && idx_root < right) {
+                int r = f(&n.value, idx_root, iterate_extra);
+                lazy_assert_zero(r);
+            }
+            if (idx_root + 1 < right) {
+                this->iterate_ptr_internal<iterate_extra_t, f>(
+                    left, right, n.right, idx_root + 1, iterate_extra);
             }
-            // don't need to copy up value, it's only used by this
-            // next call, and when that gets to the bottom there
-            // won't be any more recursion
-            n.weight--;
-            this->delete_internal(&n.right, 0, &n, rebalance_subtree);
-        }
-    } else {
-        n.weight--;
-        if (*rebalance_subtree == nullptr && this->will_need_rebalance(*subtreep, 0, -1)) {
-            *rebalance_subtree = subtreep;
-        }
-        this->delete_internal(&n.right, idx - leftweight - 1, copyn, rebalance_subtree);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_internal_array(const uint32_t left, const uint32_t right,
-                                                         iterate_extra_t *const iterate_extra) const {
-    int r;
-    for (uint32_t i = left; i < right; ++i) {
-        r = f(this->d.a.values[this->d.a.start_idx + i], i, iterate_extra);
-        if (r != 0) {
-            return r;
         }
     }
-    return 0;
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(omtdata_t *, const uint32_t, iterate_extra_t *const)>
-void omt<omtdata_t, omtdataout_t, supports_marks>::iterate_ptr_internal(const uint32_t left, const uint32_t right,
-                                                        const subtree &subtree, const uint32_t idx,
-                                                        iterate_extra_t *const iterate_extra) {
-    if (!subtree.is_null()) { 
-        omt_node &n = this->d.t.nodes[subtree.get_index()];
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename iterate_extra_t,
+              int (*f)(omtdata_t *, const uint32_t, iterate_extra_t *const)>
+    void
+    omt<omtdata_t, omtdataout_t, supports_marks>::iterate_ptr_internal_array(
+        const uint32_t left,
+        const uint32_t right,
+        iterate_extra_t *const iterate_extra) {
+        for (uint32_t i = left; i < right; ++i) {
+            int r =
+                f(&this->d.a.values[this->d.a.start_idx + i], i, iterate_extra);
+            lazy_assert_zero(r);
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_internal(
+        const uint32_t left,
+        const uint32_t right,
+        const subtree &st,
+        const uint32_t idx,
+        iterate_extra_t *const iterate_extra) const {
+        if (st.is_null()) {
+            return 0;
+        }
+        int r;
+        const omt_node &n = this->d.t.nodes[st.get_index()];
         const uint32_t idx_root = idx + this->nweight(n.left);
         if (left < idx_root) {
-            this->iterate_ptr_internal<iterate_extra_t, f>(left, right, n.left, idx, iterate_extra);
+            r = this->iterate_internal<iterate_extra_t, f>(
+                left, right, n.left, idx, iterate_extra);
+            if (r != 0) {
+                return r;
+            }
         }
         if (left <= idx_root && idx_root < right) {
-            int r = f(&n.value, idx_root, iterate_extra);
-            lazy_assert_zero(r);
+            r = f(n.value, idx_root, iterate_extra);
+            if (r != 0) {
+                return r;
+            }
         }
         if (idx_root + 1 < right) {
-            this->iterate_ptr_internal<iterate_extra_t, f>(left, right, n.right, idx_root + 1, iterate_extra);
-        }
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(omtdata_t *, const uint32_t, iterate_extra_t *const)>
-void omt<omtdata_t, omtdataout_t, supports_marks>::iterate_ptr_internal_array(const uint32_t left, const uint32_t right,
-                                                              iterate_extra_t *const iterate_extra) {
-    for (uint32_t i = left; i < right; ++i) {
-        int r = f(&this->d.a.values[this->d.a.start_idx + i], i, iterate_extra);
-        lazy_assert_zero(r);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_internal(const uint32_t left, const uint32_t right,
-                                                   const subtree &subtree, const uint32_t idx,
-                                                   iterate_extra_t *const iterate_extra) const {
-    if (subtree.is_null()) { return 0; }
-    int r;
-    const omt_node &n = this->d.t.nodes[subtree.get_index()];
-    const uint32_t idx_root = idx + this->nweight(n.left);
-    if (left < idx_root) {
-        r = this->iterate_internal<iterate_extra_t, f>(left, right, n.left, idx, iterate_extra);
-        if (r != 0) { return r; }
-    }
-    if (left <= idx_root && idx_root < right) {
-        r = f(n.value, idx_root, iterate_extra);
-        if (r != 0) { return r; }
-    }
-    if (idx_root + 1 < right) {
-        return this->iterate_internal<iterate_extra_t, f>(left, right, n.right, idx_root + 1, iterate_extra);
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_and_mark_range_internal(const uint32_t left, const uint32_t right,
-                                                                                  const subtree &subtree, const uint32_t idx,
-                                                                                  iterate_extra_t *const iterate_extra) {
-    paranoid_invariant(!subtree.is_null());
-    int r;
-    omt_node &n = this->d.t.nodes[subtree.get_index()];
-    const uint32_t idx_root = idx + this->nweight(n.left);
-    if (left < idx_root && !n.left.is_null()) {
-        n.set_marks_below_bit();
-        r = this->iterate_and_mark_range_internal<iterate_extra_t, f>(left, right, n.left, idx, iterate_extra);
-        if (r != 0) { return r; }
-    }
-    if (left <= idx_root && idx_root < right) {
-        n.set_marked_bit();
-        r = f(n.value, idx_root, iterate_extra);
-        if (r != 0) { return r; }
-    }
-    if (idx_root + 1 < right && !n.right.is_null()) {
-        n.set_marks_below_bit();
-        return this->iterate_and_mark_range_internal<iterate_extra_t, f>(left, right, n.right, idx_root + 1, iterate_extra);
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename iterate_extra_t,
-         int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::iterate_over_marked_internal(const subtree &subtree, const uint32_t idx,
-                                                                               iterate_extra_t *const iterate_extra) const {
-    if (subtree.is_null()) { return 0; }
-    int r;
-    const omt_node &n = this->d.t.nodes[subtree.get_index()];
-    const uint32_t idx_root = idx + this->nweight(n.left);
-    if (n.get_marks_below()) {
-        r = this->iterate_over_marked_internal<iterate_extra_t, f>(n.left, idx, iterate_extra);
-        if (r != 0) { return r; }
-    }
-    if (n.get_marked()) {
-        r = f(n.value, idx_root, iterate_extra);
-        if (r != 0) { return r; }
-    }
-    if (n.get_marks_below()) {
-        return this->iterate_over_marked_internal<iterate_extra_t, f>(n.right, idx_root + 1, iterate_extra);
-    }
-    return 0;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::fetch_internal_array(const uint32_t i, omtdataout_t *const value) const {
-    if (value != nullptr) {
-        copyout(value, &this->d.a.values[this->d.a.start_idx + i]);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::fetch_internal(const subtree &subtree, const uint32_t i, omtdataout_t *const value) const {
-    omt_node &n = this->d.t.nodes[subtree.get_index()];
-    const uint32_t leftweight = this->nweight(n.left);
-    if (i < leftweight) {
-        this->fetch_internal(n.left, i, value);
-    } else if (i == leftweight) {
-        if (value != nullptr) {
-            copyout(value, &n);
-        }
-    } else {
-        this->fetch_internal(n.right, i - leftweight - 1, value);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::fill_array_with_subtree_idxs(node_idx *const array, const subtree &subtree) const {
-    if (!subtree.is_null()) {
-        const omt_node &tree = this->d.t.nodes[subtree.get_index()];
-        this->fill_array_with_subtree_idxs(&array[0], tree.left);
-        array[this->nweight(tree.left)] = subtree.get_index();
-        this->fill_array_with_subtree_idxs(&array[this->nweight(tree.left) + 1], tree.right);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::rebuild_subtree_from_idxs(subtree *const subtree, const node_idx *const idxs, const uint32_t numvalues) {
-    if (numvalues==0) {
-        subtree->set_to_null();
-    } else {
-        uint32_t halfway = numvalues/2;
-        subtree->set_index(idxs[halfway]);
-        //node_idx newidx = idxs[halfway];
-        omt_node &newnode = this->d.t.nodes[subtree->get_index()];
-        newnode.weight = numvalues;
-        // value is already in there.
-        this->rebuild_subtree_from_idxs(&newnode.left,  &idxs[0], halfway);
-        this->rebuild_subtree_from_idxs(&newnode.right, &idxs[halfway+1], numvalues-(halfway+1));
-        //n_idx = newidx;
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::rebalance(subtree *const subtree) {
-    node_idx idx = subtree->get_index();
-    if (idx==this->d.t.root.get_index()) {
-        //Try to convert to an array.
-        //If this fails, (malloc) nothing will have changed.
-        //In the failure case we continue on to the standard rebalance
-        //algorithm.
-        this->convert_to_array();
-        if (supports_marks) {
-            this->convert_to_tree();
+            return this->iterate_internal<iterate_extra_t, f>(
+                left, right, n.right, idx_root + 1, iterate_extra);
         }
-    } else {
-        const omt_node &n = this->d.t.nodes[idx];
-        node_idx *tmp_array;
-        size_t mem_needed = n.weight * (sizeof tmp_array[0]);
-        size_t mem_free = (this->capacity - this->d.t.free_idx) * (sizeof this->d.t.nodes[0]);
-        bool malloced;
-        if (mem_needed<=mem_free) {
-            //There is sufficient free space at the end of the nodes array
-            //to hold enough node indexes to rebalance.
-            malloced = false;
-            tmp_array = reinterpret_cast<node_idx *>(&this->d.t.nodes[this->d.t.free_idx]);
-        }
-        else {
-            malloced = true;
-            XMALLOC_N(n.weight, tmp_array);
-        }
-        this->fill_array_with_subtree_idxs(tmp_array, *subtree);
-        this->rebuild_subtree_from_idxs(subtree, tmp_array, n.weight);
-        if (malloced) toku_free(tmp_array);
-    }
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(omtdata_t *const out, const omt_node *const n) {
-    *out = n->value;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(omtdata_t **const out, omt_node *const n) {
-    *out = &n->value;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(omtdata_t *const out, const omtdata_t *const stored_value_ptr) {
-    *out = *stored_value_ptr;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(omtdata_t **const out, omtdata_t *const stored_value_ptr) {
-    *out = stored_value_ptr;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_zero_array(const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    paranoid_invariant_notnull(idxp);
-    uint32_t min = this->d.a.start_idx;
-    uint32_t limit = this->d.a.start_idx + this->d.a.num_values;
-    uint32_t best_pos = subtree::NODE_NULL;
-    uint32_t best_zero = subtree::NODE_NULL;
-
-    while (min!=limit) {
-        uint32_t mid = (min + limit) / 2;
-        int hv = h(this->d.a.values[mid], extra);
-        if (hv<0) {
-            min = mid+1;
-        }
-        else if (hv>0) {
-            best_pos  = mid;
-            limit     = mid;
-        }
-        else {
-            best_zero = mid;
-            limit     = mid;
-        }
-    }
-    if (best_zero!=subtree::NODE_NULL) {
-        //Found a zero
-        if (value != nullptr) {
-            copyout(value, &this->d.a.values[best_zero]);
+        return 0;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::
+        iterate_and_mark_range_internal(const uint32_t left,
+                                        const uint32_t right,
+                                        const subtree &st,
+                                        const uint32_t idx,
+                                        iterate_extra_t *const iterate_extra) {
+        paranoid_invariant(!st.is_null());
+        int r;
+        omt_node &n = this->d.t.nodes[st.get_index()];
+        const uint32_t idx_root = idx + this->nweight(n.left);
+        if (left < idx_root && !n.left.is_null()) {
+            n.set_marks_below_bit();
+            r = this->iterate_and_mark_range_internal<iterate_extra_t, f>(
+                left, right, n.left, idx, iterate_extra);
+            if (r != 0) {
+                return r;
+            }
+        }
+        if (left <= idx_root && idx_root < right) {
+            n.set_marked_bit();
+            r = f(n.value, idx_root, iterate_extra);
+            if (r != 0) {
+                return r;
+            }
+        }
+        if (idx_root + 1 < right && !n.right.is_null()) {
+            n.set_marks_below_bit();
+            return this->iterate_and_mark_range_internal<iterate_extra_t, f>(
+                left, right, n.right, idx_root + 1, iterate_extra);
         }
-        *idxp = best_zero - this->d.a.start_idx;
         return 0;
     }
-    if (best_pos!=subtree::NODE_NULL) *idxp = best_pos - this->d.a.start_idx;
-    else                     *idxp = this->d.a.num_values;
-    return DB_NOTFOUND;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_zero(const subtree &subtree, const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    paranoid_invariant_notnull(idxp);
-    if (subtree.is_null()) {
-        *idxp = 0;
-        return DB_NOTFOUND;
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <
+        typename iterate_extra_t,
+        int (*f)(const omtdata_t &, const uint32_t, iterate_extra_t *const)>
+    int
+    omt<omtdata_t, omtdataout_t, supports_marks>::iterate_over_marked_internal(
+        const subtree &st,
+        const uint32_t idx,
+        iterate_extra_t *const iterate_extra) const {
+        if (st.is_null()) {
+            return 0;
+        }
+        int r;
+        const omt_node &n = this->d.t.nodes[st.get_index()];
+        const uint32_t idx_root = idx + this->nweight(n.left);
+        if (n.get_marks_below()) {
+            r = this->iterate_over_marked_internal<iterate_extra_t, f>(
+                n.left, idx, iterate_extra);
+            if (r != 0) {
+                return r;
+            }
+        }
+        if (n.get_marked()) {
+            r = f(n.value, idx_root, iterate_extra);
+            if (r != 0) {
+                return r;
+            }
+        }
+        if (n.get_marks_below()) {
+            return this->iterate_over_marked_internal<iterate_extra_t, f>(
+                n.right, idx_root + 1, iterate_extra);
+        }
+        return 0;
     }
-    omt_node &n = this->d.t.nodes[subtree.get_index()];
-    int hv = h(n.value, extra);
-    if (hv<0) {
-        int r = this->find_internal_zero<omtcmp_t, h>(n.right, extra, value, idxp);
-        *idxp += this->nweight(n.left)+1;
-        return r;
-    } else if (hv>0) {
-        return this->find_internal_zero<omtcmp_t, h>(n.left, extra, value, idxp);
-    } else {
-        int r = this->find_internal_zero<omtcmp_t, h>(n.left, extra, value, idxp);
-        if (r==DB_NOTFOUND) {
-            *idxp = this->nweight(n.left);
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::fetch_internal_array(
+        const uint32_t i,
+        omtdataout_t *const value) const {
+        if (value != nullptr) {
+            copyout(value, &this->d.a.values[this->d.a.start_idx + i]);
+        }
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::fetch_internal(
+        const subtree &st,
+        const uint32_t i,
+        omtdataout_t *const value) const {
+        omt_node &n = this->d.t.nodes[st.get_index()];
+        const uint32_t leftweight = this->nweight(n.left);
+        if (i < leftweight) {
+            this->fetch_internal(n.left, i, value);
+        } else if (i == leftweight) {
             if (value != nullptr) {
                 copyout(value, &n);
             }
-            r = 0;
+        } else {
+            this->fetch_internal(n.right, i - leftweight - 1, value);
         }
-        return r;
     }
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_plus_array(const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    paranoid_invariant_notnull(idxp);
-    uint32_t min = this->d.a.start_idx;
-    uint32_t limit = this->d.a.start_idx + this->d.a.num_values;
-    uint32_t best = subtree::NODE_NULL;
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void
+    omt<omtdata_t, omtdataout_t, supports_marks>::fill_array_with_subtree_idxs(
+        node_idx *const array,
+        const subtree &st) const {
+        if (!st.is_null()) {
+            const omt_node &tree = this->d.t.nodes[st.get_index()];
+            this->fill_array_with_subtree_idxs(&array[0], tree.left);
+            array[this->nweight(tree.left)] = st.get_index();
+            this->fill_array_with_subtree_idxs(
+                &array[this->nweight(tree.left) + 1], tree.right);
+        }
+    }
 
-    while (min != limit) {
-        const uint32_t mid = (min + limit) / 2;
-        const int hv = h(this->d.a.values[mid], extra);
-        if (hv > 0) {
-            best = mid;
-            limit = mid;
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void
+    omt<omtdata_t, omtdataout_t, supports_marks>::rebuild_subtree_from_idxs(
+        subtree *const st,
+        const node_idx *const idxs,
+        const uint32_t numvalues) {
+        if (numvalues == 0) {
+            st->set_to_null();
         } else {
-            min = mid + 1;
+            uint32_t halfway = numvalues / 2;
+            st->set_index(idxs[halfway]);
+            // node_idx newidx = idxs[halfway];
+            omt_node &newnode = this->d.t.nodes[st->get_index()];
+            newnode.weight = numvalues;
+            // value is already in there.
+            this->rebuild_subtree_from_idxs(&newnode.left, &idxs[0], halfway);
+            this->rebuild_subtree_from_idxs(
+                &newnode.right, &idxs[halfway + 1], numvalues - (halfway + 1));
+            // n_idx = newidx;
         }
     }
-    if (best == subtree::NODE_NULL) { return DB_NOTFOUND; }
-    if (value != nullptr) {
-        copyout(value, &this->d.a.values[best]);
-    }
-    *idxp = best - this->d.a.start_idx;
-    return 0;
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_plus(const subtree &subtree, const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    paranoid_invariant_notnull(idxp);
-    if (subtree.is_null()) {
-        return DB_NOTFOUND;
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::rebalance(
+        subtree *const st) {
+        node_idx idx = st->get_index();
+        if (idx == this->d.t.root.get_index()) {
+            // Try to convert to an array.
+            // If this fails, (malloc) nothing will have changed.
+            // In the failure case we continue on to the standard rebalance
+            // algorithm.
+            this->convert_to_array();
+            if (supports_marks) {
+                this->convert_to_tree();
+            }
+        } else {
+            const omt_node &n = this->d.t.nodes[idx];
+            node_idx *tmp_array;
+            size_t mem_needed = n.weight * (sizeof tmp_array[0]);
+            size_t mem_free = (this->capacity - this->d.t.free_idx) *
+                              (sizeof this->d.t.nodes[0]);
+            bool malloced;
+            if (mem_needed <= mem_free) {
+                // There is sufficient free space at the end of the nodes array
+                // to hold enough node indexes to rebalance.
+                malloced = false;
+                tmp_array = reinterpret_cast<node_idx *>(
+                    &this->d.t.nodes[this->d.t.free_idx]);
+            } else {
+                malloced = true;
+                XMALLOC_N(n.weight, tmp_array);
+            }
+            this->fill_array_with_subtree_idxs(tmp_array, *st);
+            this->rebuild_subtree_from_idxs(st, tmp_array, n.weight);
+            if (malloced)
+                toku_free(tmp_array);
+        }
     }
-    omt_node *const n = &this->d.t.nodes[subtree.get_index()];
-    int hv = h(n->value, extra);
-    int r;
-    if (hv > 0) {
-        r = this->find_internal_plus<omtcmp_t, h>(n->left, extra, value, idxp);
-        if (r == DB_NOTFOUND) {
-            *idxp = this->nweight(n->left);
-            if (value != nullptr) {
-                copyout(value, n);
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(
+        omtdata_t *const out,
+        const omt_node *const n) {
+        *out = n->value;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(
+        omtdata_t **const out,
+        omt_node *const n) {
+        *out = &n->value;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(
+        omtdata_t *const out,
+        const omtdata_t *const stored_value_ptr) {
+        *out = *stored_value_ptr;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    void omt<omtdata_t, omtdataout_t, supports_marks>::copyout(
+        omtdata_t **const out,
+        omtdata_t *const stored_value_ptr) {
+        *out = stored_value_ptr;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_zero_array(
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        paranoid_invariant_notnull(idxp);
+        uint32_t min = this->d.a.start_idx;
+        uint32_t limit = this->d.a.start_idx + this->d.a.num_values;
+        uint32_t best_pos = subtree::NODE_NULL;
+        uint32_t best_zero = subtree::NODE_NULL;
+
+        while (min != limit) {
+            uint32_t mid = (min + limit) / 2;
+            int hv = h(this->d.a.values[mid], extra);
+            if (hv < 0) {
+                min = mid + 1;
+            } else if (hv > 0) {
+                best_pos = mid;
+                limit = mid;
+            } else {
+                best_zero = mid;
+                limit = mid;
             }
-            r = 0;
         }
-    } else {
-        r = this->find_internal_plus<omtcmp_t, h>(n->right, extra, value, idxp);
-        if (r == 0) {
-            *idxp += this->nweight(n->left) + 1;
+        if (best_zero != subtree::NODE_NULL) {
+            // Found a zero
+            if (value != nullptr) {
+                copyout(value, &this->d.a.values[best_zero]);
+            }
+            *idxp = best_zero - this->d.a.start_idx;
+            return 0;
         }
+        if (best_pos != subtree::NODE_NULL)
+            *idxp = best_pos - this->d.a.start_idx;
+        else
+            *idxp = this->d.a.num_values;
+        return DB_NOTFOUND;
     }
-    return r;
-}
-
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_minus_array(const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    paranoid_invariant_notnull(idxp);
-    uint32_t min = this->d.a.start_idx;
-    uint32_t limit = this->d.a.start_idx + this->d.a.num_values;
-    uint32_t best = subtree::NODE_NULL;
 
-    while (min != limit) {
-        const uint32_t mid = (min + limit) / 2;
-        const int hv = h(this->d.a.values[mid], extra);
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_zero(
+        const subtree &st,
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        paranoid_invariant_notnull(idxp);
+        if (st.is_null()) {
+            *idxp = 0;
+            return DB_NOTFOUND;
+        }
+        omt_node &n = this->d.t.nodes[st.get_index()];
+        int hv = h(n.value, extra);
         if (hv < 0) {
-            best = mid;
-            min = mid + 1;
+            int r = this->find_internal_zero<omtcmp_t, h>(
+                n.right, extra, value, idxp);
+            *idxp += this->nweight(n.left) + 1;
+            return r;
+        } else if (hv > 0) {
+            return this->find_internal_zero<omtcmp_t, h>(
+                n.left, extra, value, idxp);
         } else {
-            limit = mid;
+            int r = this->find_internal_zero<omtcmp_t, h>(
+                n.left, extra, value, idxp);
+            if (r == DB_NOTFOUND) {
+                *idxp = this->nweight(n.left);
+                if (value != nullptr) {
+                    copyout(value, &n);
+                }
+                r = 0;
+            }
+            return r;
         }
     }
-    if (best == subtree::NODE_NULL) { return DB_NOTFOUND; }
-    if (value != nullptr) {
-        copyout(value, &this->d.a.values[best]);
-    }
-    *idxp = best - this->d.a.start_idx;
-    return 0;
-}
 
-template<typename omtdata_t, typename omtdataout_t, bool supports_marks>
-template<typename omtcmp_t,
-         int (*h)(const omtdata_t &, const omtcmp_t &)>
-int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_minus(const subtree &subtree, const omtcmp_t &extra, omtdataout_t *const value, uint32_t *const idxp) const {
-    paranoid_invariant_notnull(idxp);
-    if (subtree.is_null()) {
-        return DB_NOTFOUND;
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_plus_array(
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        paranoid_invariant_notnull(idxp);
+        uint32_t min = this->d.a.start_idx;
+        uint32_t limit = this->d.a.start_idx + this->d.a.num_values;
+        uint32_t best = subtree::NODE_NULL;
+
+        while (min != limit) {
+            const uint32_t mid = (min + limit) / 2;
+            const int hv = h(this->d.a.values[mid], extra);
+            if (hv > 0) {
+                best = mid;
+                limit = mid;
+            } else {
+                min = mid + 1;
+            }
+        }
+        if (best == subtree::NODE_NULL) {
+            return DB_NOTFOUND;
+        }
+        if (value != nullptr) {
+            copyout(value, &this->d.a.values[best]);
+        }
+        *idxp = best - this->d.a.start_idx;
+        return 0;
     }
-    omt_node *const n = &this->d.t.nodes[subtree.get_index()];
-    int hv = h(n->value, extra);
-    if (hv < 0) {
-        int r = this->find_internal_minus<omtcmp_t, h>(n->right, extra, value, idxp);
-        if (r == 0) {
-            *idxp += this->nweight(n->left) + 1;
-        } else if (r == DB_NOTFOUND) {
-            *idxp = this->nweight(n->left);
-            if (value != nullptr) {
-                copyout(value, n);
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_plus(
+        const subtree &st,
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        paranoid_invariant_notnull(idxp);
+        if (st.is_null()) {
+            return DB_NOTFOUND;
+        }
+        omt_node *const n = &this->d.t.nodes[st.get_index()];
+        int hv = h(n->value, extra);
+        int r;
+        if (hv > 0) {
+            r = this->find_internal_plus<omtcmp_t, h>(
+                n->left, extra, value, idxp);
+            if (r == DB_NOTFOUND) {
+                *idxp = this->nweight(n->left);
+                if (value != nullptr) {
+                    copyout(value, n);
+                }
+                r = 0;
+            }
+        } else {
+            r = this->find_internal_plus<omtcmp_t, h>(
+                n->right, extra, value, idxp);
+            if (r == 0) {
+                *idxp += this->nweight(n->left) + 1;
             }
-            r = 0;
         }
         return r;
-    } else {
-        return this->find_internal_minus<omtcmp_t, h>(n->left, extra, value, idxp);
     }
-}
-} // namespace toku
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_minus_array(
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        paranoid_invariant_notnull(idxp);
+        uint32_t min = this->d.a.start_idx;
+        uint32_t limit = this->d.a.start_idx + this->d.a.num_values;
+        uint32_t best = subtree::NODE_NULL;
+
+        while (min != limit) {
+            const uint32_t mid = (min + limit) / 2;
+            const int hv = h(this->d.a.values[mid], extra);
+            if (hv < 0) {
+                best = mid;
+                min = mid + 1;
+            } else {
+                limit = mid;
+            }
+        }
+        if (best == subtree::NODE_NULL) {
+            return DB_NOTFOUND;
+        }
+        if (value != nullptr) {
+            copyout(value, &this->d.a.values[best]);
+        }
+        *idxp = best - this->d.a.start_idx;
+        return 0;
+    }
+
+    template <typename omtdata_t, typename omtdataout_t, bool supports_marks>
+    template <typename omtcmp_t, int (*h)(const omtdata_t &, const omtcmp_t &)>
+    int omt<omtdata_t, omtdataout_t, supports_marks>::find_internal_minus(
+        const subtree &st,
+        const omtcmp_t &extra,
+        omtdataout_t *const value,
+        uint32_t *const idxp) const {
+        paranoid_invariant_notnull(idxp);
+        if (st.is_null()) {
+            return DB_NOTFOUND;
+        }
+        omt_node *const n = &this->d.t.nodes[st.get_index()];
+        int hv = h(n->value, extra);
+        if (hv < 0) {
+            int r = this->find_internal_minus<omtcmp_t, h>(
+                n->right, extra, value, idxp);
+            if (r == 0) {
+                *idxp += this->nweight(n->left) + 1;
+            } else if (r == DB_NOTFOUND) {
+                *idxp = this->nweight(n->left);
+                if (value != nullptr) {
+                    copyout(value, n);
+                }
+                r = 0;
+            }
+            return r;
+        } else {
+            return this->find_internal_minus<omtcmp_t, h>(
+                n->left, extra, value, idxp);
+        }
+    }
+}  // namespace toku
diff --git a/storage/tokudb/PerconaFT/util/omt.h b/storage/tokudb/PerconaFT/util/omt.h
index c7ed2ca546f..dc2fd9b7162 100644
--- a/storage/tokudb/PerconaFT/util/omt.h
+++ b/storage/tokudb/PerconaFT/util/omt.h
@@ -32,6 +32,19 @@ Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
 
     You should have received a copy of the GNU Affero General Public License
     along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
 ======= */
 
 #ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
diff --git a/storage/tokudb/ha_tokudb.cc b/storage/tokudb/ha_tokudb.cc
index 548ac5c7b09..babbe825f2e 100644
--- a/storage/tokudb/ha_tokudb.cc
+++ b/storage/tokudb/ha_tokudb.cc
@@ -7252,6 +7252,16 @@ int ha_tokudb::create(
     tokudb_trx_data *trx = NULL;
     THD* thd = ha_thd();
 
+    String database_name, table_name, dictionary_name;
+    tokudb_split_dname(name, database_name, table_name, dictionary_name);
+    if (database_name.is_empty() || table_name.is_empty()) {
+        push_warning_printf(thd,
+                            Sql_condition::WARN_LEVEL_WARN,
+                            ER_TABLE_NAME,
+                            "TokuDB: Table Name or Database Name is empty");
+        DBUG_RETURN(ER_TABLE_NAME);
+    }
+
     memset(&kc_info, 0, sizeof(kc_info));
 
 #if 100000 <= MYSQL_VERSION_ID && MYSQL_VERSION_ID <= 100999
diff --git a/storage/tokudb/hatoku_hton.cc b/storage/tokudb/hatoku_hton.cc
index 610c9e07be0..a1d6597e33a 100644
--- a/storage/tokudb/hatoku_hton.cc
+++ b/storage/tokudb/hatoku_hton.cc
@@ -575,10 +575,10 @@ static int tokudb_init_func(void *p) {
 
     db_env->set_update(db_env, tokudb_update_fun);
 
-    db_env_set_direct_io(tokudb::sysvars::directio == TRUE);
+    db_env_set_direct_io(tokudb::sysvars::directio);
 
     db_env_set_compress_buffers_before_eviction(
-        tokudb::sysvars::compress_buffers_before_eviction == TRUE);
+        tokudb::sysvars::compress_buffers_before_eviction);
 
     db_env->change_fsync_log_period(db_env, tokudb::sysvars::fsync_log_period);
 
diff --git a/storage/tokudb/hatoku_hton.h b/storage/tokudb/hatoku_hton.h
index c5b6aab1769..e90af067b00 100644
--- a/storage/tokudb/hatoku_hton.h
+++ b/storage/tokudb/hatoku_hton.h
@@ -190,7 +190,6 @@ inline bool tokudb_killed_thd_callback(void* extra,
     return thd_killed(thd) != 0;
 }
 
-extern HASH tokudb_open_tables;
 extern const char* tokudb_hton_name;
 extern int tokudb_hton_initialized;
 extern tokudb::thread::rwlock_t tokudb_hton_initialized_lock;
diff --git a/storage/tokudb/mysql-test/tokudb_bugs/r/PS-4979.result b/storage/tokudb/mysql-test/tokudb_bugs/r/PS-4979.result
new file mode 100644
index 00000000000..f0d2f93f630
--- /dev/null
+++ b/storage/tokudb/mysql-test/tokudb_bugs/r/PS-4979.result
@@ -0,0 +1,2 @@
+CREATE TABLE `#mysql50#q.q`(f1 INT KEY) ENGINE=TOKUDB;
+ERROR HY000: Got error 1632 from storage engine
diff --git a/storage/tokudb/mysql-test/tokudb_bugs/t/PS-4979.test b/storage/tokudb/mysql-test/tokudb_bugs/t/PS-4979.test
new file mode 100644
index 00000000000..1e4b5d11922
--- /dev/null
+++ b/storage/tokudb/mysql-test/tokudb_bugs/t/PS-4979.test
@@ -0,0 +1,12 @@
+--source include/have_tokudb.inc
+# PS-4979 : Dropping TokuDB table with non-alphanumeric characters could lead
+#           to a crash
+#
+# `#mysql50#q.q` is an invalid table name, but the server side doesn't detect it
+# and complain.  Instead it passes in an empty table name to the engine.  The
+# engine expects a table name in the form of a relative path like
+# "./databasename/tablename".  InnoDB detects this in parsing the table name
+# during the creation and returns an error.
+
+--error ER_GET_ERRNO
+CREATE TABLE `#mysql50#q.q`(f1 INT KEY) ENGINE=TOKUDB;
diff --git a/storage/tokudb/tokudb_background.cc b/storage/tokudb/tokudb_background.cc
index 13e0e9321cc..19f03dbca65 100644
--- a/storage/tokudb/tokudb_background.cc
+++ b/storage/tokudb/tokudb_background.cc
@@ -182,14 +182,14 @@ void* job_manager_t::real_thread_func() {
         if (res == tokudb::thread::semaphore_t::E_INTERRUPTED || _shutdown) {
                 break;
         } else if (res == tokudb::thread::semaphore_t::E_SIGNALLED) {
-#if TOKUDB_DEBUG
+#if defined(TOKUDB_DEBUG)
             if (TOKUDB_UNLIKELY(
                     tokudb::sysvars::debug_pause_background_job_manager)) {
                 _sem.signal();
                 tokudb::time::sleep_microsec(250000);
                 continue;
             }
-#endif  // TOKUDB_DEBUG
+#endif  // defined(TOKUDB_DEBUG)
 
             mutex_t_lock(_mutex);
             assert_debug(_background_jobs.size() > 0);
diff --git a/storage/tokudb/tokudb_sysvars.cc b/storage/tokudb/tokudb_sysvars.cc
index e8e9f908275..d1f58d012ec 100644
--- a/storage/tokudb/tokudb_sysvars.cc
+++ b/storage/tokudb/tokudb_sysvars.cc
@@ -662,13 +662,13 @@ static MYSQL_THDVAR_ULONGLONG(
     ~0ULL,
     1);
 
-static MYSQL_THDVAR_STR(
-    last_lock_timeout,
-    PLUGIN_VAR_MEMALLOC,
-    "last lock timeout",
-    NULL,
-    NULL,
-    NULL);
+static MYSQL_THDVAR_STR(last_lock_timeout,
+                        PLUGIN_VAR_MEMALLOC | PLUGIN_VAR_NOCMDOPT |
+                            PLUGIN_VAR_READONLY,
+                        "last lock timeout",
+                        NULL,
+                        NULL,
+                        NULL);
 
 static MYSQL_THDVAR_BOOL(
     load_save_space,
diff --git a/storage/tokudb/tokudb_sysvars.h b/storage/tokudb/tokudb_sysvars.h
index d81d5fd7999..6f09f296a80 100644
--- a/storage/tokudb/tokudb_sysvars.h
+++ b/storage/tokudb/tokudb_sysvars.h
@@ -93,10 +93,10 @@ extern my_bool      gdb_on_fatal;
 
 extern my_bool         check_jemalloc;
 
-#if TOKUDB_DEBUG
+#if defined(TOKUDB_DEBUG)
 // used to control background job manager
 extern my_bool      debug_pause_background_job_manager;
-#endif // TOKUDB_DEBUG
+#endif // defined(TOKUDB_DEBUG)
 
 // session/thread
 my_bool     alter_print_error(THD* thd);