Fix dense fragment domains during global order write with maximum fragment size (#5655)

Fixes CORE-290.

## Report

In CORE-290 a customer reported issues with corrupt arrays after running
consolidation. The symptom was memory allocation errors when opening an
array. The root cause turned out to be the consolidation itself was
writing new fragments where the fragment domain did not match the number
of tiles in the fragment.

## Root cause analysis

Stepping through a minimal reproducer revealed that consolidation is not
at fault at all - instead it is the global order writer
`max_fragment_size_` field. This field, presumably meant to be used for
consolidation only but valid for other writes, instructs the writer to
write multiple fragments each under the requested size as necessary.
When a write splits its data into multiple fragments this way, nothing
need be done for sparse arrays. But dense fragment metadata relies on
the domain to determine the number of tiles, and the global order writer
did not update the fragment metadata domain when splitting its write
into multiple fragments.

This pull request fixes that issue by ensuring that the global order
writer updates the fragment metadata domain to reflect what was actually
written into that fragment.

This is easier said than done. In CORE-290 I offered four ways we can
fix this. The chosen solution:

## Design

The fragment metadata domain is a bounding rectangle. This means that
the global order writer must split the tiles of its input into fragments
at tile boundaries which bisect the bounding rectangle into two smaller
bounding rectangles.

To do so, we add a first pass `identify_fragment_tile_boundaries` which
returns a list of tile offsets where new fragments will begin. Upon
finishing a fragment, we use that tile offset to determine which
rectangle within the target subarray the fragment actually represents,
and update the fragment metadata accordingly. We use new functions
`is_rectangular_domain` to determine whether a `(start_tile, num_tiles)`
pair identifies a rectangle, and `domain_tile_offset` to compute that
rectangle.

Much of the complexity comes from the usage of the global order writer
which does happen in consolidation: multi-part writes. A user (or a
consolidation operation) can set a domain `D` which it intends to write
into, and then actually fill in all of the cells over multiple `submit`
calls which stream in the cells to write. It is not required for these
cells to be tile aligned. Because of that, and the need to write
rectangle fragments, a single `submit` cannot always determine whether a
tail of tiles belongs to its current fragment or must be deferred to the
next. To get around this we keep those tiles in memory in the
`global_write_state_` and prepend them to the user input in the next
`submit`.

## Testing

We add unit tests for `is_rectangular_domain` and `domain_tile_offset`,
including both example tests as well as rapidcheck properties to assert
general claims.

We add tests for the global order writer to make broad claims about what
the writer is supposed to do with respect to the `max_fragment_size`
parameter. We add examples and a rapidcheck test to exercise these
claims. In particular:
- no fragment should be created larger than this size
- two adjacent fragments must exceed that size (otherwise they could be
one fragment)
- fragments correctly populate the domain of the write query

Since the original report originated from consolidation, we also add
some consolidation tests **(TODO)** mimicking the customer input.

## TODO
- [x] consolidation test which mimics the customer report
- [x] test the scenario in the `FIXME` comment in
`global_order_writer.cc`
- [x] test with variable-length attributes
- [x] more careful test of `last_tiles_` buffering which observes the
amount of data buffered for a 3D write

---
TYPE: BUG
DESC: Fix dense global order writer use of `max_fragment_size_`

Author: rroelke

test/src/unit-cppapi-consolidation.cc

@@ -3713,19 +3713,7 @@ void CSparseGlobalOrderFx::run_execute(Instance& instance) {
     ASSERTER(cursor_cells + num_cells <= expect.size());
 
     // accumulate
-    std::apply(
-        [&](auto&... field) {
-          std::apply(
-              [&](auto&... field_cursor) {
-                std::apply(
-                    [&](const auto&... field_size) {
-                      (field.accumulate_cursor(field_cursor, field_size), ...);
-                    },
-                    field_sizes);
-              },
-              outcursor);
-        },
-        std::tuple_cat(outdims, outatts));
+    templates::query::accumulate_cursor(out, outcursor, field_sizes);
 
     if (status == TILEDB_COMPLETED) {
       break;
@@ -3735,15 +3723,7 @@ void CSparseGlobalOrderFx::run_execute(Instance& instance) {
   // Clean up.
   tiledb_query_free(&query);
 
-  std::apply(
-      [outcursor](auto&... outfield) {
-        std::apply(
-            [&](const auto&... field_cursor) {
-              (outfield.finish_multipart_read(field_cursor), ...);
-            },
-            outcursor);
-      },
-      std::tuple_cat(outdims, outatts));
+  templates::query::resize(out, outcursor);
 
   ASSERTER(expect.dimensions() == outdims);
 

test/src/unit-cppapi-max-fragment-size.cc

@@ -36,7 +36,8 @@ list(APPEND TILEDB_CORE_INCLUDE_DIR "${CMAKE_SOURCE_DIR}/tiledb/sm/c_api")
 
 # Gather the test source files
 set(TILEDB_TEST_SUPPORT_SOURCES
-  rapidcheck/show.cc
+  rapidcheck/show/array_schema_templates.cc
+  rapidcheck/show/query_ast.cc
   src/array_helpers.cc
   src/array_schema_helpers.cc
   src/ast_helpers.h

test/src/unit-sparse-global-order-reader.cc

@@ -0,0 +1,198 @@
+/**
+ * @file test/support/rapidcheck/array_schema_templates.h
+ *
+ * @section LICENSE
+ *
+ * The MIT License
+ *
+ * @copyright Copyright (c) 2025 TileDB, Inc.
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ *
+ * @section DESCRIPTION
+ *
+ * This file defines rapidcheck generators for the structures
+ * defined in test/support/src/array_schema_templates.h.
+ */
+
+#ifndef TILEDB_RAPIDCHECK_ARRAY_SCHEMA_H
+#define TILEDB_RAPIDCHECK_ARRAY_SCHEMA_H
+
+#include <test/support/src/array_schema_templates.h>
+#include <test/support/stdx/tuple.h>
+#include <test/support/tdb_rapidcheck.h>
+
+#include "tiledb/common/arithmetic.h"
+
+namespace rc {
+
+using namespace tiledb::test;
+using namespace tiledb::test::templates;
+
+template <DimensionType D>
+Gen<templates::Domain<D>> make_domain(std::optional<D> bound = std::nullopt) {
+  auto bounds = gen::mapcat(gen::arbitrary<D>(), [bound](D lb) {
+    const D ub_limit =
+        (bound.has_value() ?
+             tiledb::common::checked_arithmetic<D>::add(lb, bound.value())
+                 .value_or(std::numeric_limits<D>::max()) :
+             std::numeric_limits<D>::max());
+    if constexpr (std::is_same_v<D, int64_t> || std::is_same_v<D, uint64_t>) {
+      return gen::pair(gen::just(lb), gen::inRange(lb, ub_limit));
+    } else {
+      // NB: `gen::inRange` is exclusive at the upper end but tiledb domain is
+      // inclusive. So we have to use `int64_t` to avoid overflow.
+      return gen::pair(
+          gen::just(lb),
+          gen::cast<D>(gen::inRange(int64_t(lb), int64_t(ub_limit) + 1)));
+    }
+  });
+
+  return gen::map(bounds, [](std::pair<D, D> bounds) {
+    return templates::Domain<D>(bounds.first, bounds.second);
+  });
+}
+
+template <DimensionType D>
+struct Arbitrary<templates::Domain<D>> {
+  static Gen<templates::Domain<D>> arbitrary() {
+    return make_domain<D>();
+  }
+};
+
+template <DimensionType D>
+Gen<D> make_extent(
+    const templates::Domain<D>& domain, std::optional<D> bound = std::nullopt) {
+  // upper bound on all possible extents to avoid unreasonably
+  // huge tile sizes
+  static constexpr D extent_limit = static_cast<D>(
+      std::is_signed<D>::value ?
+          std::min(
+              static_cast<int64_t>(std::numeric_limits<D>::max()),
+              static_cast<int64_t>(1024 * 16)) :
+          std::min(
+              static_cast<uint64_t>(std::numeric_limits<D>::max()),
+              static_cast<uint64_t>(1024 * 16)));
+
+  const D extent_bound =
+      (bound.has_value() ? std::min(bound.value(), extent_limit) :
+                           extent_limit);
+
+  // NB: `gen::inRange` is exclusive at the upper end but tiledb domain is
+  // inclusive. So we have to be careful to avoid overflow.
+
+  D extent_lower_bound = 1;
+  D extent_upper_bound;
+
+  const auto bound_distance = tiledb::common::checked_arithmetic<D>::sub(
+      domain.upper_bound, domain.lower_bound);
+  if (bound_distance.has_value()) {
+    extent_upper_bound =
+        (bound_distance.value() < extent_bound ? bound_distance.value() + 1 :
+                                                 extent_bound);
+  } else {
+    extent_upper_bound = extent_bound;
+  }
+
+  return gen::inRange(extent_lower_bound, extent_upper_bound + 1);
+}
+
+template <tiledb::sm::Datatype D>
+Gen<templates::Dimension<D>> make_dimension(
+    std::optional<typename templates::Dimension<D>::value_type> extent_bound =
+        std::nullopt,
+    std::optional<typename templates::Dimension<D>::value_type> domain_bound =
+        std::nullopt) {
+  using CoordType = templates::Dimension<D>::value_type;
+  auto tup = gen::mapcat(
+      make_domain<CoordType>(domain_bound),
+      [extent_bound](Domain<CoordType> domain) {
+        return gen::pair(gen::just(domain), make_extent(domain, extent_bound));
+      });
+
+  return gen::map(tup, [](std::pair<Domain<CoordType>, CoordType> tup) {
+    return templates::Dimension<D>(tup.first, tup.second);
+  });
+}
+
+template <tiledb::sm::Datatype D>
+struct Arbitrary<templates::Dimension<D>> {
+  static Gen<templates::Dimension<D>> arbitrary() {
+    return make_dimension<D>();
+  }
+};
+
+template <DimensionType D>
+Gen<D> make_coordinate(const templates::Domain<D>& domain) {
+  // `gen::inRange` does an exclusive upper bound,
+  // whereas the domain upper bound is inclusive.
+  // As a result some contortion is required to deal
+  // with numeric_limits.
+  if constexpr (std::is_same_v<D, StringDimensionCoordType>) {
+    // NB: poor performance with small domains for sure
+    return gen::suchThat(
+        gen::map(
+            gen::string<std::string>(),
+            [](std::string s) {
+              StringDimensionCoordType v(s.begin(), s.end());
+              return v;
+            }),
+        [domain](const StringDimensionCoordType& s) {
+          return domain.lower_bound <= s && s <= domain.upper_bound;
+        });
+  } else if constexpr (std::is_signed<D>::value) {
+    if (int64_t(domain.upper_bound) < std::numeric_limits<int64_t>::max()) {
+      return gen::cast<D>(gen::inRange(
+          int64_t(domain.lower_bound), int64_t(domain.upper_bound + 1)));
+    } else {
+      return gen::inRange(domain.lower_bound, domain.upper_bound);
+    }
+  } else {
+    if (uint64_t(domain.upper_bound) < std::numeric_limits<uint64_t>::max()) {
+      return gen::cast<D>(gen::inRange(
+          uint64_t(domain.lower_bound), uint64_t(domain.upper_bound + 1)));
+    } else {
+      return gen::inRange(domain.lower_bound, domain.upper_bound);
+    }
+  }
+}
+
+template <DimensionType D>
+Gen<templates::Domain<D>> make_range(const templates::Domain<D>& domain) {
+  return gen::apply(
+      [](D p1, D p2) { return templates::Domain<D>(p1, p2); },
+      make_coordinate<D>(domain),
+      make_coordinate<D>(domain));
+}
+
+template <>
+void show<Domain<int>>(const templates::Domain<int>& domain, std::ostream& os);
+
+template <>
+void show<Domain<uint64_t>>(
+    const templates::Domain<uint64_t>& domain, std::ostream& os);
+
+template <>
+void show<Dimension<tiledb::sm::Datatype::UINT64>>(
+    const templates::Dimension<tiledb::sm::Datatype::UINT64>& dimension,
+    std::ostream& os);
+
+}  // namespace rc
+
+#endif