SC-61223: fix sparse global order reader to do "repeatable reads" when there are multiple fragments with the same timestamp (#5459)

When multiple fragments have the same timestamp, the sparse global order
reader (and likely others) do not have "repeatable reads". That is, if
you run the same read query multiple times, then you do not receive the
same result.

The story provides a reproducer which does something like:
```
for i in range(0, 10):
    write a new fragment A[0] = i at timestamp t=1
    for j in range(0, 10):
        read A[0]
```

Below is the result:
```
python tmp/write_consistency.py 
OrderedDict([('a', array([], dtype=int32)), ('d1', array([], dtype=int64))])
Wrote 0, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 1, read [0, 1, 1, 1, 1, 1, 1, 1, 1, 0]
Wrote 2, read [0, 1, 0, 0, 1, 2, 1, 0, 1, 1]
Wrote 3, read [0, 0, 0, 0, 3, 2, 2, 1, 2, 1]
Wrote 4, read [0, 3, 4, 1, 4, 2, 1, 2, 1, 4]
Wrote 5, read [1, 1, 5, 4, 3, 0, 0, 4, 0, 0]
Wrote 6, read [4, 6, 0, 2, 0, 3, 4, 4, 4, 4]
Wrote 7, read [4, 4, 4, 4, 7, 4, 5, 4, 4, 7]
Wrote 8, read [5, 0, 0, 7, 7, 0, 0, 4, 4, 4]
Wrote 9, read [8, 4, 3, 5, 2, 7, 9, 8, 2, 7]
```
We can see that the reads do not all produce the same value.

In the story we discussed whether we would be satisfied with "repeatable
read", in which each row from the reproducing script contains just one
distinct value; or whether we need "strict write order", in which each
row `i` contains only values `i`.

Repeatable read is sufficient, so that's what is implemented in this
pull request. An example new result of the above script is:
```
OrderedDict([('a', array([], dtype=int32)), ('d1', array([], dtype=int64))])
Wrote 0, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 1, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 2, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 3, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 4, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 5, read [0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
Wrote 6, read [6, 6, 6, 6, 6, 6, 6, 6, 6, 6]
Wrote 7, read [6, 6, 6, 6, 6, 6, 6, 6, 6, 6]
Wrote 8, read [6, 6, 6, 6, 6, 6, 6, 6, 6, 6]
Wrote 9, read [6, 6, 6, 6, 6, 6, 6, 6, 6, 6]
```

# Implementation

The implementation ultimately is straightforward. The coordinates
comparator breaks ties in coordinate values using the tile timestamp. We
add an additional tiebreaker which compares the fragment index. The
fragment index is itself determined by the lexicographic ordering of
fragment names, which includes the timestamp and UUID; as such for a
fixed array the fragment indices will always be the same for each read.
The greatest fragment index in typical cases is the most recently
written, so it wins the comparison.

("In typical cases"... meaning what? If two fragments are written at the
same timestamp which is `now()`, then there is logic in the UUID
generation to ensure that the first fragment's UUID is lexicographically
less than the second fragment's UUID. However, this logic is incorrect
if the array was opened at a fixed non-`now()` timestamp. Multiple
fragments written at a fixed timestamp will have uncorrelated UUIDs and
thus can appear in any order when reading the array. Hence we get a
repeatable read order but not always one which matches the physical
order of the writes..)

In any case, I spent a bit of time diving into the above, and there are
some artifacts which I elected to leave in here because they are both
harmless and tested - specifically some additional accessors to
components of `FragmentID`.

---
TYPE: BUG
DESC: repeatable read for multiple fragments written at fixed timestamp

Author: rroelke

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

@@ -461,7 +461,7 @@ struct CSparseGlobalOrderFx {
       int* coords, uint64_t* coords_size, int* data, uint64_t* data_size);
 
   template <typename Asserter, FragmentType Fragment>
-  void write_fragment(const Fragment& fragment);
+  void write_fragment(const Fragment& fragment, CApiArray* existing = nullptr);
 
   void write_1d_fragment_strings(
       int* coords,
@@ -630,9 +630,16 @@ void CSparseGlobalOrderFx::write_1d_fragment(
  * Writes a generic `FragmentType` into the array.
  */
 template <typename Asserter, FragmentType Fragment>
-void CSparseGlobalOrderFx::write_fragment(const Fragment& fragment) {
-  // Open array for writing.
-  CApiArray array(context(), array_name_.c_str(), TILEDB_WRITE);
+void CSparseGlobalOrderFx::write_fragment(
+    const Fragment& fragment, CApiArray* existing) {
+  std::optional<CApiArray> constructed;
+  if (!existing) {
+    // Open array for writing.
+    constructed.emplace(context(), array_name_.c_str(), TILEDB_WRITE);
+    existing = &constructed.value();
+  }
+
+  CApiArray& array = *existing;
 
   const auto dimensions = fragment.dimensions();
   const auto attributes = fragment.attributes();
@@ -772,6 +779,15 @@ int32_t CSparseGlobalOrderFx::read(
   // Open array for reading.
   CApiArray array(context(), array_name_.c_str(), TILEDB_READ);
 
+  const char* dimname = array->array()
+                            ->array_schema_latest()
+                            .domain()
+                            .dimension_ptr(0)
+                            ->name()
+                            .c_str();
+  const char* attname =
+      array->array()->array_schema_latest().attribute(0)->name().c_str();
+
   // Create query.
   tiledb_query_t* query;
   auto rc = tiledb_query_alloc(context(), array, TILEDB_READ, &query);
@@ -797,7 +813,12 @@ int32_t CSparseGlobalOrderFx::read(
     if (qc_idx == 1) {
       int32_t val = 11;
       rc = tiledb_query_condition_init(
-          context(), query_condition, "a", &val, sizeof(int32_t), TILEDB_LT);
+          context(),
+          query_condition,
+          attname,
+          &val,
+          sizeof(int32_t),
+          TILEDB_LT);
       CHECK(rc == TILEDB_OK);
     } else if (qc_idx == 2) {
       // Negated query condition should produce the same results.
@@ -806,7 +827,7 @@ int32_t CSparseGlobalOrderFx::read(
       rc = tiledb_query_condition_alloc(context(), &qc);
       CHECK(rc == TILEDB_OK);
       rc = tiledb_query_condition_init(
-          context(), qc, "a", &val, sizeof(int32_t), TILEDB_GE);
+          context(), qc, attname, &val, sizeof(int32_t), TILEDB_GE);
       CHECK(rc == TILEDB_OK);
       rc = tiledb_query_condition_negate(context(), qc, &query_condition);
       CHECK(rc == TILEDB_OK);
@@ -822,9 +843,10 @@ int32_t CSparseGlobalOrderFx::read(
 
   rc = tiledb_query_set_layout(context(), query, TILEDB_GLOBAL_ORDER);
   CHECK(rc == TILEDB_OK);
-  rc = tiledb_query_set_data_buffer(context(), query, "a", data, data_size);
+  rc = tiledb_query_set_data_buffer(context(), query, attname, data, data_size);
   CHECK(rc == TILEDB_OK);
-  rc = tiledb_query_set_data_buffer(context(), query, "d", coords, coords_size);
+  rc = tiledb_query_set_data_buffer(
+      context(), query, dimname, coords, coords_size);
   CHECK(rc == TILEDB_OK);
 
   // Submit query.
@@ -2968,6 +2990,147 @@ TEST_CASE_METHOD(
   tiledb_query_free(&query);
 }
 
+/**
+ * Test that we get "repeatable reads" when multiple fragments
+ * are written at the same timestamp.  That is, for a fixed array
+ * A, reading the contents of A should always produce the same results.
+ */
+TEST_CASE_METHOD(
+    CSparseGlobalOrderFx,
+    "Sparse global order reader: repeatable reads for sub-millisecond "
+    "fragments",
+    "[sparse-global-order][sub-millisecond][rest][rapidcheck]") {
+  auto doit = [this]<typename Asserter>(
+                  const std::vector<uint64_t> fragment_same_timestamp_runs) {
+    uint64_t num_fragments = 0;
+    for (const auto same_timestamp_run : fragment_same_timestamp_runs) {
+      num_fragments += same_timestamp_run;
+    }
+
+    FxRun2D instance;
+    instance.allow_dups = false;
+
+    /*
+     * Each fragment (T, F) writes its fragment index into both (1, 1)
+     * and (2 + T, 2 + F).
+     *
+     * (1, 1) is the coordinate where they must be de-duplicated.
+     * The other coordinate is useful for ensuring that all fragments
+     * are included in the result set, and for debugging by inspecting tile
+     * MBRs.
+     */
+    for (uint64_t t = 0; t < fragment_same_timestamp_runs.size(); t++) {
+      for (uint64_t f = 0; f < fragment_same_timestamp_runs[t]; f++) {
+        FxRun2D::FragmentType fragment;
+        fragment.d1_ = {1, 2 + static_cast<int>(t)};
+        fragment.d2_ = {1, 2 + static_cast<int>(f)};
+        std::get<0>(fragment.atts_) = {
+            static_cast<int>(instance.fragments.size()),
+            static_cast<int>(instance.fragments.size())};
+
+        instance.fragments.push_back(fragment);
+      }
+    }
+
+    create_array<Asserter, decltype(instance)>(instance);
+
+    DeleteArrayGuard arrayguard(context(), array_name_.c_str());
+
+    /*
+     * Write each fragment at a fixed timestamp.
+     * Opening for write at timestamp `t` causes all the fragments to have `t`
+     * as their start and end timestamps.
+     */
+    for (uint64_t i = 0, t = 0; t < fragment_same_timestamp_runs.size(); t++) {
+      tiledb_array_t* raw_array;
+      TRY(context(),
+          tiledb_array_alloc(context(), array_name_.c_str(), &raw_array));
+      TRY(context(),
+          tiledb_array_set_open_timestamp_start(context(), raw_array, t + 1));
+      TRY(context(),
+          tiledb_array_set_open_timestamp_end(context(), raw_array, t + 1));
+
+      CApiArray array(context(), raw_array, TILEDB_WRITE);
+      for (uint64_t f = 0; f < fragment_same_timestamp_runs[t]; f++, i++) {
+        write_fragment<Asserter, decltype(instance.fragments[i])>(
+            instance.fragments[i], &array);
+      }
+    }
+
+    CApiArray array(context(), array_name_.c_str(), TILEDB_READ);
+
+    // Value from (1, 1).
+    // Because all the writes are at the same timestamp we make no guarantee
+    // of ordering. `attvalue` may be the value from any of the fragments,
+    // but it must be the same value each time we read.
+    std::optional<int> attvalue;
+
+    for (uint64_t f = 0; f < num_fragments; f++) {
+      std::vector<int> dim1(instance.fragments.size() * 4);
+      std::vector<int> dim2(instance.fragments.size() * 4);
+      std::vector<int> atts(instance.fragments.size() * 4);
+      uint64_t dim1_size = sizeof(int) * dim1.size();
+      uint64_t dim2_size = sizeof(int) * dim2.size();
+      uint64_t atts_size = sizeof(int) * atts.size();
+
+      tiledb_query_t* query;
+      TRY(context(), tiledb_query_alloc(context(), array, TILEDB_READ, &query));
+      TRY(context(),
+          tiledb_query_set_layout(context(), query, TILEDB_GLOBAL_ORDER));
+      TRY(context(),
+          tiledb_query_set_data_buffer(
+              context(), query, "d1", dim1.data(), &dim1_size));
+      TRY(context(),
+          tiledb_query_set_data_buffer(
+              context(), query, "d2", dim2.data(), &dim2_size));
+      TRY(context(),
+          tiledb_query_set_data_buffer(
+              context(), query, "a1", atts.data(), &atts_size));
+
+      tiledb_query_status_t status;
+      TRY(context(), tiledb_query_submit(context(), query));
+      TRY(context(), tiledb_query_get_status(context(), query, &status));
+      tiledb_query_free(&query);
+      ASSERTER(status == TILEDB_COMPLETED);
+
+      ASSERTER(dim1_size == (1 + num_fragments) * sizeof(int));
+      ASSERTER(dim2_size == (1 + num_fragments) * sizeof(int));
+      ASSERTER(atts_size == (1 + num_fragments) * sizeof(int));
+
+      ASSERTER(dim1[0] == 1);
+      ASSERTER(dim2[0] == 1);
+      if (attvalue.has_value()) {
+        ASSERTER(attvalue.value() == atts[0]);
+      } else {
+        attvalue.emplace(atts[0]);
+      }
+
+      uint64_t c = 1;
+      for (uint64_t t = 0; t < fragment_same_timestamp_runs.size(); t++) {
+        for (uint64_t f = 0; f < fragment_same_timestamp_runs[t]; f++, c++) {
+          ASSERTER(dim1[c] == 2 + static_cast<int>(t));
+          ASSERTER(dim2[c] == 2 + static_cast<int>(f));
+          ASSERTER(atts[c] == static_cast<int>(c - 1));
+        }
+      }
+    }
+  };
+
+  SECTION("Example") {
+    doit.operator()<AsserterCatch>({10});
+  }
+
+  SECTION("Rapidcheck") {
+    rc::prop("rapidcheck consistent read order for sub-millisecond", [doit]() {
+      const auto runs = *rc::gen::suchThat(
+          rc::gen::nonEmpty(rc::gen::container<std::vector<uint64_t>>(
+              rc::gen::inRange(1, 8))),
+          [](auto value) { return value.size() <= 6; });
+      doit.operator()<AsserterRapidcheck>(runs);
+    });
+  }
+}
+
 /**
  * Creates an array with a schema whose dimensions and attributes
  * come from `Instance`.

test/support/src/array_helpers.h

@@ -101,6 +101,12 @@ struct CApiArray {
     throw_if_error(ctx, tiledb_array_open(ctx, array_, mode));
   }
 
+  CApiArray(tiledb_ctx_t* ctx, tiledb_array_t* array, tiledb_query_type_t mode)
+      : ctx_(ctx)
+      , array_(array) {
+    throw_if_error(ctx, tiledb_array_open(ctx, array_, mode));
+  }
+
   CApiArray(CApiArray&& from)
       : ctx_(from.ctx_)
       , array_(from.movefrom()) {

tiledb/sm/fragment/fragment_identifier.h

@@ -72,6 +72,12 @@ enum class FragmentNameVersion { ONE, TWO, THREE };
  */
 class FragmentID : private URI {
  private:
+  /**
+   * Whitebox testing class provides additional accessors to components of the
+   * fragment name.
+   */
+  friend class WhiteboxFragmentID;
+
   /** The fragment name. */
   std::string name_;
   /** The timestamp range. */
@@ -94,7 +100,7 @@ class FragmentID : private URI {
   ~FragmentID() = default;
 
   /** Accessor to the fragment name. */
-  inline const std::string& name() {
+  inline const std::string& name() const {
     return name_;
   }
 
@@ -103,7 +109,7 @@ class FragmentID : private URI {
    * For array format version <= 2, only the range start is valid
    * (the range end is ignored).
    */
-  inline timestamp_range_type timestamp_range() {
+  inline timestamp_range_type timestamp_range() const {
     return timestamp_range_;
   }
 
@@ -118,7 +124,7 @@ class FragmentID : private URI {
    * - Name version 3 corresponds to format version 5 or higher
    *      * __t1_t2_uuid_version
    */
-  inline int name_version() {
+  inline int name_version() const {
     if (name_version_ == FragmentNameVersion::ONE) {
       return 1;
     } else if (name_version_ == FragmentNameVersion::TWO) {
@@ -132,7 +138,7 @@ class FragmentID : private URI {
    * Accessor to the array format version.
    * Returns UINT32_MAX for name versions <= 2.
    */
-  inline format_version_t array_format_version() {
+  inline format_version_t array_format_version() const {
     return array_format_version_;
   }
 };