diff --git a/core/src/main/clojure/xtdb/operator/scan.clj b/core/src/main/clojure/xtdb/operator/scan.clj
index 5f20c577d8..4daf54a68d 100644
--- a/core/src/main/clojure/xtdb/operator/scan.clj
+++ b/core/src/main/clojure/xtdb/operator/scan.clj
@@ -15,19 +15,18 @@
             [xtdb.vector.reader :as vr]
             [xtdb.vector.writer :as vw]
             xtdb.watermark)
-  (:import (clojure.lang IFn IPersistentMap MapEntry)
-           (java.util ArrayList Arrays Iterator LinkedList List Map)
+  (:import (clojure.lang IPersistentMap MapEntry )
+           (java.util ArrayList Arrays Iterator LinkedList List Map TreeSet NavigableSet ListIterator)
            (java.util.function IntConsumer)
            org.apache.arrow.memory.ArrowBuf
            org.apache.arrow.memory.BufferAllocator
            [org.apache.arrow.memory.util ArrowBufPointer]
-           (org.apache.arrow.vector BigIntVector NullVector VectorLoader VectorSchemaRoot)
-           (org.apache.arrow.vector.complex ListVector StructVector)
+           (org.apache.arrow.vector VectorLoader VectorSchemaRoot)
            xtdb.api.protocols.TransactionInstant
            xtdb.buffer_pool.IBufferPool
            xtdb.ICursor
            xtdb.indexer.live_index.ILiveTableWatermark
-           (xtdb.metadata IMetadataManager ITableMetadata)
+           (xtdb.metadata IMetadataManager)
            xtdb.object_store.ObjectStore
            xtdb.operator.IRelationSelector
            (xtdb.trie ArrowHashTrie ArrowHashTrie$Leaf HashTrie LeafMerge LeafMerge$LeafPointer LiveHashTrie$Leaf)
@@ -169,9 +168,6 @@
         (apply-constraint for-system-time "xt$system_from" "xt$system_to")))
     range))
 
-(defn- scan-op-at-now [scan-op]
-  (= :now (first (second scan-op))))
-
 (defn- scan-op-point? [scan-op]
   (= :at (first scan-op)))
 
@@ -181,298 +177,30 @@
        (or (nil? for-system-time)
            (scan-op-point? for-system-time))))
 
-(defn- range-point-query? [^IWatermark watermark basis {:keys [for-system-time] :as _scan-opts}]
-  (and
-   (.txBasis watermark)
-   (= (:tx basis)
-      (.txBasis watermark))
-   (or (nil? for-system-time)
-       (scan-op-point? for-system-time))
-   (>= (util/instant->micros (:current-time basis))
-       (util/instant->micros (:system-time (:tx basis))))))
+(defn- at-range-point? [{:keys [for-system-time]}]
+  (or (nil? for-system-time)
+      (scan-op-point? for-system-time)))
+
 (defn tables-with-cols [basis ^IWatermarkSource wm-src ^IScanEmitter scan-emitter]
   (let [{:keys [tx, after-tx]} basis
         wm-tx (or tx after-tx)]
     (with-open [^Watermark wm (.openWatermark wm-src wm-tx)]
       (.allTableColNames scan-emitter wm))))
 
-(defn- point-point-row-picker
-  ^java.util.function.IntConsumer [^IRelationWriter out-rel, ^RelationReader leaf-rel
-                                   col-names, ^longs temporal-range,
-                                   {:keys [^longs current-bounds skip-iid-ptr prev-iid-ptr current-iid-ptr]}]
-
-  (let [leaf-row-count (.rowCount leaf-rel)
-        iid-rdr (.readerForName leaf-rel "xt$iid")
-        sys-from-rdr (.readerForName leaf-rel "xt$system_from")
-        op-rdr (.readerForName leaf-rel "op")
-        put-rdr (.legReader op-rdr :put)
-        doc-rdr (.structKeyReader put-rdr "xt$doc")
-        put-valid-from-rdr (.structKeyReader put-rdr "xt$valid_from")
-        put-valid-to-rdr (.structKeyReader put-rdr "xt$valid_to")
-
-        delete-rdr (.legReader op-rdr :delete)
-        delete-valid-from-rdr (.structKeyReader delete-rdr "xt$valid_from")
-        delete-valid-to-rdr (.structKeyReader delete-rdr "xt$valid_to")
-
-        row-copiers (vec
-                     (for [col-name col-names
-                           :let [normalized-name (util/str->normal-form-str col-name)
-                                 ^IVectorReader rdr (case normalized-name
-                                                      "xt$iid" (.readerForName leaf-rel "xt$iid")
-                                                      "xt$system_from" sys-from-rdr
-                                                      "xt$system_to" (vr/vec->reader
-                                                                      (doto (NullVector. "xt$system_to")
-                                                                        (.setValueCount leaf-row-count)))
-                                                      "xt$valid_from" nil
-                                                      "xt$valid_to" nil
-                                                      (.structKeyReader doc-rdr normalized-name))]
-                           :when rdr]
-                       (.rowCopier rdr
-                                   (case normalized-name
-                                     "xt$iid" (.writerForName out-rel col-name [:fixed-size-binary 16])
-                                     "xt$system_from" (.writerForName out-rel col-name types/temporal-col-type)
-                                     "xt$system_to" (.writerForName out-rel col-name
-                                                                    [:union #{:null types/temporal-col-type}])
-                                     (.writerForName out-rel col-name)))))
-
-        valid-from-wtrs (vec
-                         (for [col-name col-names
-                               :when (= "xt$valid_from" (util/str->normal-form-str col-name))]
-                           (.writerForName out-rel col-name types/temporal-col-type)))
-
-        valid-to-wtrs (vec
-                       (for [col-name col-names
-                             :when (= "xt$valid_to" (util/str->normal-form-str col-name))]
-                         (.writerForName out-rel col-name types/temporal-col-type)))
-
-        valid-time (aget temporal-range valid-from-upper-idx)
-        system-time (aget temporal-range system-from-upper-idx)]
-
-    (letfn [(duplicate-ptr [^ArrowBufPointer dst, ^ArrowBufPointer src]
-              (.set dst (.getBuf src) (.getOffset src) (.getLength src)))
-
-            (constrain-bounds [^long valid-from, ^long valid-to]
-              (when (and (< valid-to valid-time)
-                         (< (aget current-bounds 0) valid-to))
-                (aset current-bounds 0 valid-to))
-
-              (when (and (> valid-from valid-time)
-                         (< valid-from (aget current-bounds 1)))
-                (aset current-bounds 1 valid-from)))]
-
-      (reify IntConsumer
-        (accept [_ idx]
-          (when-not (= skip-iid-ptr (.getPointer iid-rdr idx current-iid-ptr))
-            (when-not (= prev-iid-ptr current-iid-ptr)
-              (aset current-bounds 0 Long/MIN_VALUE)
-              (aset current-bounds 1 Long/MAX_VALUE)
-              (duplicate-ptr prev-iid-ptr current-iid-ptr))
+;; As the algorithm processes events in reverse system time order, one can
+;; immediately write out the system-to times when having finished an event.
+;; The system-to times are not relevant for processing earlier events.
+(deftype Rectangle [^long valid-from, ^long valid-to, ^long sys-from])
 
-            (if (= :evict (.getLeg op-rdr idx))
-              (duplicate-ptr skip-iid-ptr current-iid-ptr)
-              (when (<= (.getLong sys-from-rdr idx) system-time)
-                (case (.getLeg op-rdr idx)
-                  :put
-                  (let [valid-from (.getLong put-valid-from-rdr idx)
-                        valid-to (.getLong put-valid-to-rdr idx)]
-                    (if (and (<= valid-from valid-time) (< valid-time valid-to))
-                      (do
-                        (duplicate-ptr skip-iid-ptr current-iid-ptr)
-                        (.startRow out-rel)
-                        (doseq [^IRowCopier copier row-copiers]
-                          (.copyRow copier idx))
-                        (doseq [^IVectorWriter valid-from-wtr valid-from-wtrs]
-                          (.writeLong valid-from-wtr (Long/max valid-from (aget current-bounds 0))))
-                        (doseq [^IVectorWriter valid-to-wtr valid-to-wtrs]
-                          (.writeLong valid-to-wtr (Long/min valid-to (aget current-bounds 1))))
-                        (.endRow out-rel))
-
-                      (constrain-bounds valid-from valid-to)))
-
-                  :delete
-                  (let [valid-from (.getLong delete-valid-from-rdr idx)
-                        valid-to (.getLong delete-valid-to-rdr idx)]
-                    (if (and (<= valid-from valid-time) (< valid-time valid-to))
-                      (duplicate-ptr skip-iid-ptr current-iid-ptr)
-                      (constrain-bounds valid-from valid-to))))))))))))
-
-(deftype Interval [^long start, ^long end, ^long sys-from])
-
-(defn- intersect [^Interval i1 ^Interval i2]
-  (and (<= (.start i1) (.end i2)) (<= (.start i2) (.end i1))))
-
-(defn- split
-  "i1 comes before i2 in system time"
-  [^Interval i1 ^Interval i2]
-  (let [start1 (.start i1)
-        start2 (.start i2)
-        end1 (.end i1)
-        end2 (.end i2)
-        new-sys-from (.sys-from i2)]
-    (cond-> []
-      (< start1 start2)
-      (conj (Interval. start1 start2 new-sys-from))
-      (< end2 end1)
-      (conj (Interval. end2 end1 new-sys-from)))))
-
-(defn range-point-row-picker
-  ^java.util.function.IntConsumer [^IRelationWriter out-rel, ^RelationReader leaf-rel
-                                   col-names, ^longs temporal-ranges,
-                                   {:keys [^LinkedList !ranges skip-iid-ptr prev-iid-ptr current-iid-ptr]}]
-  (let [leaf-row-count (.rowCount leaf-rel)
-        iid-rdr (.readerForName leaf-rel "xt$iid")
-        sys-from-rdr (.readerForName leaf-rel "xt$system_from")
-        op-rdr (.readerForName leaf-rel "op")
-        put-rdr (.legReader op-rdr :put)
-        doc-rdr (.structKeyReader put-rdr "xt$doc")
-        put-valid-from-rdr (.structKeyReader put-rdr "xt$valid_from")
-        put-valid-to-rdr (.structKeyReader put-rdr "xt$valid_to")
-
-        delete-rdr (.legReader op-rdr :delete)
-        delete-valid-from-rdr (.structKeyReader delete-rdr "xt$valid_from")
-        delete-valid-to-rdr (.structKeyReader delete-rdr "xt$valid_to")
-
-        valid-from-lower (aget temporal-ranges valid-from-lower-idx)
-        valid-from-upper (aget temporal-ranges valid-from-upper-idx)
-        valid-to-lower (aget temporal-ranges valid-to-lower-idx)
-        valid-to-upper (aget temporal-ranges valid-to-upper-idx)
-        system-time (aget temporal-ranges system-to-lower-idx)
-
-        row-copiers (vec
-                     (for [col-name col-names
-                           :let [normalized-name (util/str->normal-form-str col-name)
-                                 ^IVectorReader rdr (case normalized-name
-                                                      "xt$iid" (.readerForName leaf-rel "xt$iid")
-                                                      "xt$system_from" nil
-                                                      "xt$system_to" (vr/vec->reader
-                                                                      (doto (NullVector. "xt$system_to")
-                                                                        (.setValueCount leaf-row-count)))
-                                                      "xt$valid_from" nil
-                                                      "xt$valid_to" nil
-                                                      (.structKeyReader doc-rdr normalized-name))]
-                           :when rdr]
-                       (.rowCopier rdr
-                                   (case normalized-name
-                                     "xt$iid" (.writerForName out-rel col-name [:fixed-size-binary 16])
-                                     "xt$system_to" (.writerForName out-rel col-name
-                                                                    [:union #{:null types/temporal-col-type}])
-                                     (.writerForName out-rel col-name)))))
-
-        valid-from-wtrs (vec
-                         (for [col-name col-names
-                               :when (= "xt$valid_from" (util/str->normal-form-str col-name))]
-                           (.writerForName out-rel col-name types/temporal-col-type)))
-
-        valid-to-wtrs (vec
-                       (for [col-name col-names
-                             :when (= "xt$valid_to" (util/str->normal-form-str col-name))]
-                         (.writerForName out-rel col-name types/temporal-col-type)))
-
-        sys-from-wtrs (vec
-                       (for [col-name col-names
-                             :when (= "xt$system_from" (util/str->normal-form-str col-name))]
-                         (.writerForName out-rel col-name types/temporal-col-type)))
-
-        !new-ranges (LinkedList.)]
-
-    (letfn [(duplicate-ptr [^ArrowBufPointer dst, ^ArrowBufPointer src]
-              (.set dst (.getBuf src) (.getOffset src) (.getLength src)))]
-
-      (reify IntConsumer
-        (accept [_ idx]
-          (when-not (= skip-iid-ptr (.getPointer iid-rdr idx current-iid-ptr))
-            (when-not (= prev-iid-ptr current-iid-ptr)
-              (.clear !ranges)
-              (duplicate-ptr prev-iid-ptr current-iid-ptr))
-
-            (if (= :evict (.getLeg op-rdr idx))
-              (duplicate-ptr skip-iid-ptr current-iid-ptr)
-              (let [sys-from (.getLong sys-from-rdr idx)]
-                (when (<= sys-from system-time)
-                  (case (.getLeg op-rdr idx)
-                    :put
-                    (let [i1 (Interval. (.getLong put-valid-from-rdr idx) (.getLong put-valid-to-rdr idx) sys-from)
-                          itr (.listIterator !ranges)]
-                      (.add !new-ranges i1)
-                      (while (.hasNext itr)
-                        (let [i2 (.next itr)
-                              inner-itr (.listIterator !new-ranges)]
-                          (while (.hasNext inner-itr)
-                            (let [i1 (.next inner-itr)]
-                              (when (intersect i1 i2)
-                                (.remove inner-itr)
-                                (run! #(.add inner-itr %) (split i1 i2)))))))
-                      (doseq [^Interval i !new-ranges]
-                        (.add !ranges i)
-                        (let [valid-from (.start i)
-                              valid-to (.end i)
-                              sys-from (.sys-from i)]
-                          (when (and (<= valid-from-lower valid-from)
-                                     (<= valid-from valid-from-upper)
-                                     (<= valid-to-lower valid-to)
-                                     (<= valid-to valid-to-upper))
-                            (.startRow out-rel)
-                            (doseq [^IRowCopier copier row-copiers]
-                              (.copyRow copier idx))
-                            (doseq [^IVectorWriter valid-from-wtr valid-from-wtrs]
-                              (.writeLong valid-from-wtr valid-from))
-                            (doseq [^IVectorWriter valid-to-wtr valid-to-wtrs]
-                              (.writeLong valid-to-wtr valid-to))
-                            (doseq [^IVectorWriter sys-from-wtr sys-from-wtrs]
-                              (.writeLong sys-from-wtr sys-from))
-                            (.endRow out-rel))))
-                      (.clear !new-ranges))
-
-                    :delete
-                    (let [i1 (Interval. (.getLong delete-valid-from-rdr idx) (.getLong delete-valid-to-rdr idx) sys-from)
-                          itr (.listIterator !ranges)]
-                      (.add !new-ranges i1)
-                      (while (.hasNext itr)
-                        (let [i2 (.next itr)
-                              inner-itr (.listIterator !new-ranges)]
-                          (while (.hasNext inner-itr)
-                            (let [i1 (.next inner-itr)]
-                              (when (intersect i1 i2)
-                                (.remove inner-itr)
-                                (run! #(.add inner-itr %) (split i1 i2)))))))
-                      (doseq [i !new-ranges]
-                        (.add !ranges i))
-                      (.clear !new-ranges))))))))))))
-
-(deftype Rectangle [^long valid-from, ^long valid-to,
-                    ^long sys-from, ^long sys-to])
-
-(defn- rectangle-intersect [^Rectangle r1 ^Rectangle r2]
-  (not (or (> (.valid-from r2) (.valid-to r1))
-           (> (.valid-from r1) (.valid-to r2))
-           (> (.sys-from r2) (.sys-to r1))
-           (> (.sys-from r1) (.sys-to r2)))))
-
-(defn- rectangle-split
-  "r1 comes before r2 in system time"
-  [^Rectangle r1 ^Rectangle r2]
-  (let [valid-from1 (.valid-from r1)
-        valid-from2 (.valid-from r2)
-        valid-to1 (.valid-to r1)
-        valid-to2 (.valid-to r2)
-        sys-from1 (.sys-from r1)
-        sys-from2 (.sys-from r2)
-        sys-to1 (.sys-to r1)
-        sys-to2 (.sys-to r2)]
-    (cond-> []
-      (< sys-from1 sys-from2)
-      (conj (Rectangle. valid-from1 valid-to1 sys-from1 sys-from2))
-      (< sys-to2 sys-to1)
-      (conj (Rectangle. valid-from1 valid-to1 sys-to2 sys-to1))
-      (< valid-from1 valid-from2)
-      (conj (Rectangle. valid-from1 valid-from2 (max sys-from1 sys-from2) (min sys-to1 sys-to2)))
-      (< valid-to2 valid-to1)
-      (conj (Rectangle. valid-to2 valid-to1 (max sys-from1 sys-from2) (min sys-to1 sys-to2))))))
+(defn- ranges-invariant [^LinkedList !ranges]
+  (every? true? (map (fn [^Rectangle r1 ^Rectangle r2] (<= (.valid-to r1) (.valid-from r2)))
+                     !ranges (rest !ranges))))
 
 (defn range-range-row-picker
   ^java.util.function.IntConsumer [^IRelationWriter out-rel, ^RelationReader leaf-rel
                                    col-names, ^longs temporal-ranges,
-                                   {:keys [^LinkedList !ranges skip-iid-ptr prev-iid-ptr current-iid-ptr]}]
+                                   {:keys [^LinkedList !ranges skip-iid-ptr prev-iid-ptr current-iid-ptr
+                                           point-point? range-range?]}]
   (let [iid-rdr (.readerForName leaf-rel "xt$iid")
         sys-from-rdr (.readerForName leaf-rel "xt$system_from")
         op-rdr (.readerForName leaf-rel "op")
@@ -530,10 +258,53 @@
                            :when (= "xt$system_to" (util/str->normal-form-str col-name))]
                        (.writerForName out-rel col-name types/temporal-col-type)))
 
-        !new-ranges (LinkedList.)]
+        !overlapping-ranges (ArrayList.)]
 
     (letfn [(duplicate-ptr [^ArrowBufPointer dst, ^ArrowBufPointer src]
-              (.set dst (.getBuf src) (.getOffset src) (.getLength src)))]
+              (.set dst (.getBuf src) (.getOffset src) (.getLength src)))
+
+            (write-r [idx valid-from valid-to sys-from sys-to]
+              (when (and (<= valid-from-lower valid-from)
+                         (<= valid-from valid-from-upper)
+                         (<= valid-to-lower valid-to)
+                         (<= valid-to valid-to-upper)
+                         (<= sys-from-lower sys-from)
+                         (<= sys-from sys-from-upper)
+                         (<= sys-to-lower sys-to)
+                         (<= sys-to sys-to-upper)
+                         (not= valid-from valid-to)
+                         (not= sys-from sys-to))
+                (when point-point? (duplicate-ptr skip-iid-ptr current-iid-ptr))
+                (.startRow out-rel)
+                (doseq [^IRowCopier copier row-copiers]
+                  (.copyRow copier idx))
+                (doseq [^IVectorWriter valid-from-wtr valid-from-wtrs]
+                  (.writeLong valid-from-wtr valid-from))
+                (doseq [^IVectorWriter valid-to-wtr valid-to-wtrs]
+                  (.writeLong valid-to-wtr valid-to))
+                (doseq [^IVectorWriter sys-from-wtr sys-from-wtrs]
+                  (.writeLong sys-from-wtr sys-from))
+                (doseq [^IVectorWriter sys-to-wtr sys-to-wtrs]
+                  (.writeLong sys-to-wtr sys-to))
+                (.endRow out-rel)))
+            (calc-overlapping-ranges [valid-from valid-to]
+              (let [^ListIterator itr (.iterator !ranges)
+                    ^Rectangle cur (loop []
+                                     (when (.hasNext itr)
+                                       (let [^Rectangle next (.next itr)]
+                                         (if (<= (.valid-to next) valid-from)
+                                           (recur)
+                                           next))))]
+                (loop [^Rectangle cur cur]
+                  (when cur
+                    (if (< (.valid-from cur) valid-to)
+                      (do
+                        (.add !overlapping-ranges cur)
+                        (.remove itr)
+                        (recur (when (.hasNext itr) (.next itr))))
+                      (when (<= valid-to (.valid-from cur))
+                        (.previous itr)))))
+                itr))]
 
       (reify IntConsumer
         (accept [_ idx]
@@ -542,74 +313,65 @@
               (.clear !ranges)
               (duplicate-ptr prev-iid-ptr current-iid-ptr))
 
+            (assert (ranges-invariant !ranges))
+
             (if (= :evict (.getLeg op-rdr idx))
               (duplicate-ptr skip-iid-ptr current-iid-ptr)
               (let [system-from (.getLong sys-from-rdr idx)]
-                ;; TODO potentially more fancy check here for skipping
+                (.clear !overlapping-ranges)
                 (when (and (<= sys-from-lower system-from) (<= system-from sys-from-upper))
                   (case (.getLeg op-rdr idx)
                     :put
-                    (let [r1 (Rectangle. (.getLong put-valid-from-rdr idx) (.getLong put-valid-to-rdr idx)
-                                         (.getLong sys-from-rdr idx) util/end-of-time-μs)
-                          itr (.listIterator !ranges)]
-                      (.add !new-ranges r1)
-                      (while (.hasNext itr)
-                        (let [r2 (.next itr)
-                              inner-itr (.listIterator !new-ranges)]
-                          (while (.hasNext inner-itr)
-                            (let [r1 (.next inner-itr)]
-                              (when (rectangle-intersect r1 r2)
-                                (.remove inner-itr)
-                                (run! #(.add inner-itr %) (rectangle-split r1 r2)))))))
-                      (doseq [^Rectangle r !new-ranges]
-                        (.add !ranges r)
-                        (let [valid-from (.valid-from r)
-                              valid-to (.valid-to r)
-                              sys-from (.sys-from r)
-                              sys-to (.sys-to r)]
-                          (when (and (<= valid-from-lower valid-from)
-                                     (<= valid-from valid-from-upper)
-                                     (<= valid-to-lower valid-to)
-                                     (<= valid-to valid-to-upper)
-                                     (<= sys-from-lower sys-from)
-                                     (<= sys-from sys-from-upper)
-                                     (<= sys-to-lower sys-to)
-                                     (<= sys-to sys-to-upper)
-                                     (not= sys-from sys-to))
-                            (.startRow out-rel)
-                            (doseq [^IRowCopier copier row-copiers]
-                              (.copyRow copier idx))
-                            (doseq [^IVectorWriter valid-from-wtr valid-from-wtrs]
-                              (.writeLong valid-from-wtr valid-from))
-                            (doseq [^IVectorWriter valid-to-wtr valid-to-wtrs]
-                              (.writeLong valid-to-wtr valid-to))
-                            (doseq [^IVectorWriter sys-from-wtr sys-from-wtrs]
-                              (.writeLong sys-from-wtr sys-from))
-                            (doseq [^IVectorWriter sys-to-wtr sys-to-wtrs]
-                              (.writeLong sys-to-wtr sys-to))
-                            (.endRow out-rel))))
-                      (.clear !new-ranges))
+                    (let [valid-from (.getLong put-valid-from-rdr idx)
+                          valid-to (.getLong put-valid-to-rdr idx)
+                          ^ListIterator itr (calc-overlapping-ranges valid-from valid-to)]
+
+                      (when-let [^Rectangle begin (first !overlapping-ranges)]
+                        (when (< (.valid-from begin) valid-from)
+                          (.add itr (Rectangle. (.valid-from begin) valid-from (.sys-from begin))))
+                        (when (< valid-from (.valid-from begin))
+                          (write-r idx valid-from (.valid-from begin) system-from util/end-of-time-μs)))
+
+                      (if (seq !overlapping-ranges)
+                        (do
+                          (dorun (map (fn [^Rectangle r1 ^Rectangle r2]
+                                        (when (< (.valid-to r1) (.valid-from r2))
+                                          (write-r idx (.valid-to r1) (.valid-from r2) system-from util/end-of-time-μs)))
+                                      !overlapping-ranges (rest !overlapping-ranges)))
+
+                          (when range-range?
+                            (doseq [^Rectangle r !overlapping-ranges]
+                              (let [new-valid-from (max valid-from (.valid-from r))
+                                    new-valid-to (min valid-to (.valid-to r))]
+                                (when (< new-valid-from new-valid-to)
+                                  (write-r idx new-valid-from new-valid-to system-from (.sys-from r)))))))
+                        (write-r idx valid-from valid-to system-from util/end-of-time-μs))
+
+                      (.add itr (Rectangle. valid-from valid-to system-from))
+
+                      (when-let [^Rectangle end (last !overlapping-ranges)]
+                        (when (< valid-to (.valid-to end))
+                          (.add itr (Rectangle. valid-to (.valid-to end) (.sys-from end))))
+                        (when (< (.valid-to end) valid-to)
+                          (write-r idx (.valid-to end) valid-to system-from util/end-of-time-μs))))
 
                     :delete
-                    (let [r1 (Rectangle. (.getLong delete-valid-from-rdr idx) (.getLong delete-valid-to-rdr idx)
-                                         (.getLong sys-from-rdr idx) util/end-of-time-μs)
-                          itr (.listIterator !ranges)]
-                      (.add !new-ranges r1)
-                      (while (.hasNext itr)
-                        (let [r2 (.next itr)
-                              inner-itr (.listIterator !new-ranges)]
-                          (while (.hasNext inner-itr)
-                            (let [r1 (.next inner-itr)]
-                              (when (rectangle-intersect r1 r2)
-                                (.remove inner-itr)
-                                (run! #(.add inner-itr %) (rectangle-split r1 r2)))))))
-                      (doseq [i !new-ranges]
-                        (.add !ranges i))
-                      (.clear !new-ranges))))))))))))
+                    (let [valid-from (.getLong delete-valid-from-rdr idx)
+                          valid-to (.getLong delete-valid-to-rdr idx)
+                          ^ListIterator itr (calc-overlapping-ranges valid-from valid-to)]
+
+                      (when-let [^Rectangle begin (first !overlapping-ranges)]
+                        (when (< (.valid-from begin) valid-from)
+                          (.add itr (Rectangle. (.valid-from begin) valid-from (.sys-from begin)))))
+
+                      (.add itr (Rectangle. valid-from valid-to system-from))
+
+                      (when-let [^Rectangle end (last !overlapping-ranges)]
+                        (when (< valid-to (.valid-to end))
+                          (.add itr (Rectangle. valid-to (.valid-to end) (.sys-from end))))))))))))))))
 
 (deftype TrieCursor [^BufferAllocator allocator, arrow-leaves
                      ^Iterator merge-tasks, ^ints leaf-idxs, ^ints current-arrow-page-idxs
-                     ^IFn row-picker
                      col-names, ^Map col-preds, ^longs temporal-timestamps,
                      params, ^IPersistentMap picker-state]
   ICursor
@@ -620,7 +382,7 @@
           (letfn [(rel->leaf-ptr [leaf-ordinal ^RelationReader log-rdr]
                     (let [row-count (.rowCount log-rdr)
                           iid-rdr (.readerForName log-rdr "xt$iid")
-                          ^IntConsumer picker (row-picker out-rel log-rdr col-names temporal-timestamps picker-state)
+                          ^IntConsumer picker (range-range-row-picker out-rel log-rdr col-names temporal-timestamps picker-state)
 
                           is-valid-buf (ArrowBufPointer.)]
 
@@ -718,12 +480,6 @@
 ;; The consumers for different leafs need to share some state so the logic of how to advance
 ;; is correct. For example if the `skip-iid-ptr` gets set in one leaf consumer it should also affect
 ;; the skipping in another leaf consumer.
-(defn- ->picker-state []
-  {:current-bounds (long-array 2)
-   :!ranges (LinkedList.)
-   :skip-iid-ptr (ArrowBufPointer.)
-   :prev-iid-ptr (ArrowBufPointer.)
-   :current-iid-ptr (ArrowBufPointer.)})
 
 (defn ->4r-cursor [^BufferAllocator allocator, ^ObjectStore obj-store, ^IBufferPool buffer-pool, ^IWatermark wm
                    table-name, col-names, ^longs temporal-range
@@ -736,19 +492,21 @@
                                  live-table-wm inc))
                     (doto (int-array (count arrow-leaves))
                       (Arrays/fill -1))
-                    (cond (at-point-point? scan-opts)
-                          point-point-row-picker
-
-                          (range-point-query? wm basis scan-opts)
-                          range-point-row-picker
-
-                          :else
-                          range-range-row-picker)
-
                     col-names col-preds
                     temporal-range
                     params
-                    (->picker-state))
+                    (merge {:!ranges (LinkedList.)
+                            :skip-iid-ptr (ArrowBufPointer.)
+                            :prev-iid-ptr (ArrowBufPointer.)
+                            :current-iid-ptr (ArrowBufPointer.)}
+                           (cond (at-point-point? scan-opts)
+                                 {:point-point? true}
+
+                                 (at-range-point? scan-opts)
+                                 {:range-point? true}
+
+                                 :else
+                                 {:range-range? true})))
 
       (catch Throwable t
         (util/close (map :leaf-buf arrow-leaves))
diff --git a/src/test/clojure/xtdb/as_of_test.clj b/src/test/clojure/xtdb/as_of_test.clj
index 94bd5803e0..ee0fd667e9 100644
--- a/src/test/clojure/xtdb/as_of_test.clj
+++ b/src/test/clojure/xtdb/as_of_test.clj
@@ -45,12 +45,12 @@
                     :xt/valid-from system-time
                     :xt/valid-to end-of-time-zdt
                     :xt/system-from system-time
-                    :xt/system-to nil}
+                    :xt/system-to (util/->zdt util/end-of-time)}
               :doc-with-app-time {:xt/id :doc-with-app-time,
                                   :xt/valid-from (util/->zdt #inst "2021")
                                   :xt/valid-to end-of-time-zdt
                                   :xt/system-from system-time
-                                  :xt/system-to nil}}
+                                  :xt/system-to (util/->zdt util/end-of-time)}}
              (->> (tu/query-ra '[:scan {:table xt_docs}
                                  [xt/id
                                   xt/valid-from xt/valid-to
@@ -67,15 +67,15 @@
 
         original-v0-doc {:xt/id :doc, :version 0
                          :xt/valid-from tt1
-                         :xt/valid-to end-of-time-zdt
+                         :xt/valid-to tt2
                          :xt/system-from tt1
-                         :xt/system-to tt2}
+                         :xt/system-to end-of-time-zdt}
 
         replaced-v0-doc {:xt/id :doc, :version 0
-                         :xt/valid-from tt1
-                         :xt/valid-to tt2
-                         :xt/system-from tt2
-                         :xt/system-to end-of-time-zdt}
+                         :xt/valid-from tt2
+                         :xt/valid-to end-of-time-zdt
+                         :xt/system-from tt1
+                         :xt/system-to tt2}
 
         v1-doc {:xt/id :doc, :version 1
                 :xt/valid-from tt2
@@ -83,9 +83,7 @@
                 :xt/system-from tt2
                 :xt/system-to end-of-time-zdt}]
 
-    ;; TODO nil system-to #2655
-    (t/is (= #{(assoc replaced-v0-doc :xt/system-to nil)
-               (assoc v1-doc :xt/system-to nil)}
+    (t/is (= #{original-v0-doc v1-doc}
              (set (tu/query-ra '[:scan {:table xt_docs}
                                  [xt/id version
                                   xt/valid-from xt/valid-to
diff --git a/src/test/clojure/xtdb/datalog_test.clj b/src/test/clojure/xtdb/datalog_test.clj
index 76844085a7..ebc4002fef 100644
--- a/src/test/clojure/xtdb/datalog_test.clj
+++ b/src/test/clojure/xtdb/datalog_test.clj
@@ -1504,13 +1504,13 @@
             "cross-time join - who was here in both 2018 and 2023?")
 
       (t/is (= #{{:vt-start (util/->zdt #inst "2021")
-                  :vt-end (util/->zdt util/end-of-time)
-                  :tt-start (util/->zdt #inst "2020-01-01")
-                  :tt-end (util/->zdt #inst "2020-01-02")}
-                 {:vt-start (util/->zdt #inst "2021")
                   :vt-end (util/->zdt #inst "2022")
-                  :tt-start (util/->zdt #inst "2020-01-02")
-                  :tt-end (util/->zdt util/end-of-time)}}
+                  :tt-start (util/->zdt #inst "2020")
+                  :tt-end (util/->zdt util/end-of-time)}
+                 {:vt-start (util/->zdt #inst "2022")
+                  :vt-end (util/->zdt util/end-of-time)
+                  :tt-start (util/->zdt #inst "2020")
+                  :tt-end (util/->zdt  #inst "2020-01-02")}}
                (set (q '{:find [vt-start vt-end tt-start tt-end]
                          :where [(match :xt_docs {:xt/id :luke
                                                   :xt/valid-from vt-start
@@ -1657,13 +1657,14 @@
                        tx1, nil)))
             "as-of 18 Jan")
 
-      (t/is (= [{:cust 145, :app-start (util/->zdt #inst "1998-01-05")}
-                {:cust 827, :app-start (util/->zdt #inst "1998-01-12")}]
-               (q '{:find [cust app-start]
-                    :where [(match :docs {:customer-number cust, :xt/valid-from app-start}
-                                   {:for-valid-time :all-time})]
-                    :order-by [[app-start :asc]]}
-                  tx5, nil))
+      (t/is (= #{{:cust 145, :app-start (util/->zdt #inst "1998-01-05")}
+                 {:cust 827, :app-start (util/->zdt #inst "1998-01-12")}}
+               (set (q '{:find [cust app-start]
+                         :where [(match :docs {:customer-number cust,
+                                               :xt/valid-from app-start}
+                                        {:for-valid-time :all-time})]
+                         :order-by [[app-start :asc]]}
+                       tx5, nil)))
             "as-of 29 Jan")
 
       (t/is (= [{:cust 827, :app-start (util/->zdt #inst "1998-01-12"), :app-end (util/->zdt #inst "1998-01-20")}]
diff --git a/src/test/clojure/xtdb/indexer_test.clj b/src/test/clojure/xtdb/indexer_test.clj
index b776f84704..f9115f3821 100644
--- a/src/test/clojure/xtdb/indexer_test.clj
+++ b/src/test/clojure/xtdb/indexer_test.clj
@@ -185,7 +185,7 @@
                  :xt/valid-from (util/->zdt tt)
                  :xt/valid-to (util/->zdt util/end-of-time)
                  :xt/system-from (util/->zdt tt)
-                 :xt/system-to nil}]
+                 :xt/system-to (util/->zdt util/end-of-time)}]
                (tu/query-ra '[:scan {:table xt_docs}
                               [xt/id version
                                xt/valid-from, xt/valid-to
@@ -195,14 +195,14 @@
       (let [{tt2 :system-time} (xt/submit-tx node [[:put :xt_docs {:xt/id :foo, :version 1}]]
                                              {:default-all-valid-time? false})]
         (t/is (= #{{:xt/id :foo, :version 0,
-                    :xt/valid-from (util/->zdt tt)
+                    :xt/valid-from (util/->zdt tt2)
                     :xt/valid-to (util/->zdt util/end-of-time)
                     :xt/system-from (util/->zdt tt)
                     :xt/system-to (util/->zdt tt2)}
                    {:xt/id :foo, :version 0,
                     :xt/valid-from (util/->zdt tt)
                     :xt/valid-to (util/->zdt tt2)
-                    :xt/system-from (util/->zdt tt2)
+                    :xt/system-from (util/->zdt tt)
                     :xt/system-to (util/->zdt util/end-of-time)}
                    {:xt/id :foo, :version 1,
                     :xt/valid-from (util/->zdt tt2)
diff --git a/src/test/clojure/xtdb/operator/scan_test.clj b/src/test/clojure/xtdb/operator/scan_test.clj
index 4a7c71ad5a..3839e1c8b4 100644
--- a/src/test/clojure/xtdb/operator/scan_test.clj
+++ b/src/test/clojure/xtdb/operator/scan_test.clj
@@ -61,41 +61,40 @@
                                   [:delete :xt_docs :doc3]])]
 
       ;; valid-time
-      (t/is (= #{{:v 1, :xt/id :doc1} {:v 1, :xt/id :doc2}}
-               (set (tu/query-ra '[:scan
-                                   {:table xt_docs, :for-valid-time [:at #inst "2017"], :for-system-time nil}
-                                   [xt/id v]]
-                                 {:node node}))))
-
-      (t/is (= #{{:v 1, :xt/id :doc2} {:v 2, :xt/id :doc1}}
-               (set (tu/query-ra '[:scan
-                                   {:table xt_docs, :for-valid-time [:at :now], :for-system-time nil}
-                                   [xt/id v]]
-                                 {:node node}))))
+      (t/is (= {{:v 1, :xt/id :doc1} 1 {:v 1, :xt/id :doc2} 1}
+               (frequencies (tu/query-ra '[:scan
+                                           {:table xt_docs, :for-valid-time [:at #inst "2017"], :for-system-time nil}
+                                           [xt/id v]]
+                                         {:node node}))))
+      (t/is (= {{:v 1, :xt/id :doc2} 1 {:v 2, :xt/id :doc1} 1}
+               (frequencies (tu/query-ra '[:scan
+                                           {:table xt_docs, :for-valid-time [:at :now], :for-system-time nil}
+                                           [xt/id v]]
+                                         {:node node}))))
       ;; system-time
-      (t/is (= #{{:v 1, :xt/id :doc1} {:v 1, :xt/id :doc2} {:v 1, :xt/id :doc3}}
-               (set (tu/query-ra '[:scan
-                                   {:table xt_docs, :for-valid-time [:at :now], :for-system-time nil}
-                                   [xt/id v]]
-                                 {:node node :basis {:tx tx1}}))))
-
-      (t/is (= #{{:v 1, :xt/id :doc1} {:v 1, :xt/id :doc2}}
-               (set (tu/query-ra '[:scan
-                                   {:table xt_docs, :for-valid-time [:at #inst "2017"], :for-system-time nil}
-                                   [xt/id v]]
-                                 {:node node :basis {:tx tx1}}))))
-
-      (t/is (= #{{:v 2, :xt/id :doc1} {:v 1, :xt/id :doc2}}
-               (set (tu/query-ra '[:scan
-                                   {:table xt_docs, :for-valid-time [:at :now], :for-system-time nil}
-                                   [xt/id v]]
-                                 {:node node :basis {:tx tx2}}))))
-
-      (t/is (= #{{:v 2, :xt/id :doc1} {:v 2, :xt/id :doc2}}
-               (set (tu/query-ra '[:scan
-                                   {:table xt_docs, :for-valid-time [:at #inst "2100"], :for-system-time nil}
-                                   [xt/id v]]
-                                 {:node node :basis {:tx tx2}})))))))
+      (t/is (= {{:v 1, :xt/id :doc1} 1 {:v 1, :xt/id :doc2} 1 {:v 1, :xt/id :doc3} 1}
+               (frequencies (tu/query-ra '[:scan
+                                           {:table xt_docs, :for-valid-time [:at :now], :for-system-time nil}
+                                           [xt/id v]]
+                                         {:node node :basis {:tx tx1}}))))
+
+      (t/is (= {{:v 1, :xt/id :doc1} 1 {:v 1, :xt/id :doc2} 1}
+               (frequencies (tu/query-ra '[:scan
+                                           {:table xt_docs, :for-valid-time [:at #inst "2017"], :for-system-time nil}
+                                           [xt/id v]]
+                                         {:node node :basis {:tx tx1}}))))
+
+      (t/is (= {{:v 2, :xt/id :doc1} 1 {:v 1, :xt/id :doc2} 1}
+               (frequencies (tu/query-ra '[:scan
+                                           {:table xt_docs, :for-valid-time [:at :now], :for-system-time nil}
+                                           [xt/id v]]
+                                         {:node node :basis {:tx tx2}}))))
+
+      (t/is (= {{:v 2, :xt/id :doc1} 1 {:v 2, :xt/id :doc2} 1}
+               (frequencies (tu/query-ra '[:scan
+                                           {:table xt_docs, :for-valid-time [:at #inst "2100"], :for-system-time nil}
+                                           [xt/id v]]
+                                         {:node node :basis {:tx tx2}})))))))
 
 (t/deftest test-past-point-point-queries-with-valid-time
   (with-open [node (node/start-node tu/*node-opts*)]
@@ -210,7 +209,7 @@
       (t/is (= #{{:xt/valid-from (util/->zdt tt)
                   :xt/valid-to (util/->zdt util/end-of-time)
                   :xt/system-from (util/->zdt tt),
-                  :xt/system-to nil}}
+                  :xt/system-to (util/->zdt util/end-of-time)}}
                (set (tu/query-ra '[:scan {:table xt_docs}
                                    [xt/valid-from xt/valid-to
                                     xt/system-from xt/system-to]]
@@ -224,18 +223,30 @@
 
     #_(tu/finish-chunk! node)
 
-    (t/is (= #{{:xt/system-from (util/->zdt #inst "3000")
-                :xt/system-to (util/->zdt #inst "3001")
-                :last_updated "tx1"}
-               {:xt/system-from (util/->zdt #inst "3001")
-                :xt/system-to (util/->zdt util/end-of-time)
-                :last_updated "tx1"}
-               {:xt/system-from (util/->zdt #inst "3001")
-                :xt/system-to (util/->zdt util/end-of-time)
-                :last_updated "tx2"}}
+    (t/is (= #{{:last_updated "tx2",
+                :xt/valid-from #time/zoned-date-time "3001-01-01T00:00Z[UTC]",
+                :xt/valid-to
+                #time/zoned-date-time "9999-12-31T23:59:59.999999Z[UTC]",
+                :xt/system-from #time/zoned-date-time "3001-01-01T00:00Z[UTC]",
+                :xt/system-to
+                #time/zoned-date-time "9999-12-31T23:59:59.999999Z[UTC]"}
+               {:last_updated "tx1",
+                :xt/valid-from #time/zoned-date-time "3000-01-01T00:00Z[UTC]",
+                :xt/valid-to #time/zoned-date-time "3001-01-01T00:00Z[UTC]",
+                :xt/system-from #time/zoned-date-time "3000-01-01T00:00Z[UTC]",
+                :xt/system-to
+                #time/zoned-date-time "9999-12-31T23:59:59.999999Z[UTC]"}
+               {:last_updated "tx1",
+                :xt/valid-from #time/zoned-date-time "3001-01-01T00:00Z[UTC]",
+                :xt/valid-to
+                #time/zoned-date-time "9999-12-31T23:59:59.999999Z[UTC]",
+                :xt/system-from #time/zoned-date-time "3000-01-01T00:00Z[UTC]",
+                :xt/system-to #time/zoned-date-time "3001-01-01T00:00Z[UTC]"}}
              (set (tu/query-ra '[:scan {:table foo, :for-system-time :all-time}
                                  [{xt/system-from (< xt/system-from #time/zoned-date-time "3002-01-01T00:00Z")}
                                   {xt/system-to (> xt/system-to #time/zoned-date-time "2999-01-01T00:00Z")}
+                                  xt/valid-from
+                                  xt/valid-to
                                   last_updated]]
                                {:node node :default-all-valid-time? true}))))))
 
@@ -371,3 +382,131 @@
              (tu/query-ra
               '[:scan {:table xt_docs} [xt/id {col-x (= col-x "toto")}]]
               {:node node})))))
+
+(t/deftest test-correct-rectangle-cutting
+  (with-open [node (node/start-node {})]
+    (letfn [(q [id]
+              (-> (xt/q node {:find '[v vf vt]
+                              :where [(list 'match :xt_docs
+                                            ['v {:xt/id id :xt/valid-from 'vf :xt/valid-to 'vt}]
+                                            {:for-valid-time :all-time})]})
+                  frequencies))]
+      (t/testing "period starts before and does NOT overlap"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 1, :v 1} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 1, :v 2} {:for-valid-time [:in #inst "2005" #inst "2009"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2005-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2009-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1}
+                 (q 1))))
+
+      (t/testing "period starts before and overlaps"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 2, :v 1} {:for-valid-time [:in #inst "2015" #inst "2025"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 2, :v 2} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2020-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2025-01-01T00:00Z[UTC]"} 1}
+                 (q 2))))
+
+      (t/testing "period starts equally and overlaps"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 3, :v 1} {:for-valid-time [:in #inst "2010" #inst "2025"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 3, :v 2} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2020-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2025-01-01T00:00Z[UTC]"} 1}
+                 (q 3))))
+
+      (t/testing "newer period completely covered"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 4, :v 1} {:for-valid-time [:in #inst "2010" #inst "2025"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 4, :v 2} {:for-valid-time [:in #inst "2015" #inst "2020"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2015-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2015-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2020-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2025-01-01T00:00Z[UTC]"} 1}
+                 (q 4))))
+
+      (t/testing "older period completely covered"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 5, :v 1} {:for-valid-time [:in #inst "2015" #inst "2020"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 5, :v 2} {:for-valid-time [:in #inst "2010" #inst "2025"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2025-01-01T00:00Z[UTC]"} 1}
+                 (q 5))))
+
+      (t/testing "period end equally and overlaps"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 6, :v 1} {:for-valid-time [:in #inst "2010" #inst "2025"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 6, :v 2} {:for-valid-time [:in #inst "2015" #inst "2025"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2015-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2025-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2015-01-01T00:00Z[UTC]"} 1}
+                 (q 6))))
+
+      (t/testing "period ends after and overlaps"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 7, :v 1} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 7, :v 2} {:for-valid-time [:in #inst "2015" #inst "2025"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2015-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2025-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2015-01-01T00:00Z[UTC]"} 1}
+                 (q 7))))
+
+      (t/testing "period starts before and touches"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 8, :v 1} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 8, :v 2} {:for-valid-time [:in #inst "2005" #inst "2010"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2005-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2010-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1}
+                 (q 8))))
+
+      (t/testing "period starts after and touches"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 9, :v 1} {:for-valid-time [:in #inst "2005" #inst "2010"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 9, :v 2} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2005-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2010-01-01T00:00Z[UTC]"} 1}
+                 (q 9))))
+
+      (t/testing "period starts after and does NOT overlap"
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 10, :v 1} {:for-valid-time [:in #inst "2005" #inst "2009"]}]])
+        (xt/submit-tx node [[:put :xt_docs {:xt/id 10, :v 2} {:for-valid-time [:in #inst "2010" #inst "2020"]}]])
+
+        (t/is (= {{:v 2,
+                   :vf #time/zoned-date-time "2010-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2020-01-01T00:00Z[UTC]"} 1
+                  {:v 1,
+                   :vf #time/zoned-date-time "2005-01-01T00:00Z[UTC]",
+                   :vt #time/zoned-date-time "2009-01-01T00:00Z[UTC]"} 1}
+                 (q 10)))))))
diff --git a/src/test/clojure/xtdb/operator_test.clj b/src/test/clojure/xtdb/operator_test.clj
index 29ab406154..b6a81a4981 100644
--- a/src/test/clojure/xtdb/operator_test.clj
+++ b/src/test/clojure/xtdb/operator_test.clj
@@ -124,7 +124,8 @@
         (t/is (= #{"tx1" "tx2"}
                  (q '{xt$system_from (<= xt$system_from ?system-time2)})))
 
-        (t/is (= #{"tx1" "tx2"}
+        ;; this test depends on how one cuts rectangles
+        (t/is (= #{"tx2"} #_#{"tx1" "tx2"}
                  (q '{xt$system_from (> xt$system_from ?system-time1)})))
 
         (t/is (= #{}
diff --git a/src/test/resources/xtdb/sql/logic_test/direct-sql/sl-demo.test b/src/test/resources/xtdb/sql/logic_test/direct-sql/sl-demo.test
index 1c2d63cfa1..5bb29a1a6e 100644
--- a/src/test/resources/xtdb/sql/logic_test/direct-sql/sl-demo.test
+++ b/src/test/resources/xtdb/sql/logic_test/direct-sql/sl-demo.test
@@ -22,7 +22,7 @@ SELECT *
 1998-01-10T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
-NULL
+9999-12-31T23:59:59.999999Z[UTC]
 
 statement ok
 INSERT INTO Prop_Owner (xt$id, customer_number, property_number, xt$valid_from)
@@ -44,7 +44,7 @@ SELECT *
 1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
-NULL
+9999-12-31T23:59:59.999999Z[UTC]
 
 query IIITTTT rowsort
 SELECT *
@@ -64,12 +64,12 @@ SELECT *
 7797
 1998-01-10T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -99,6 +99,7 @@ SELECT *
                xt$system_to)
 ----
 
+
 query IIITTTT rowsort
 SELECT *
   FROM Prop_Owner
@@ -117,19 +118,12 @@ SELECT *
 7797
 1998-01-10T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -138,12 +132,12 @@ SELECT *
 7797
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-02T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 827
 7797
-1998-01-15T00:00Z[UTC]
+1998-01-20T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
 2020-01-03T00:00Z[UTC]
@@ -177,26 +171,12 @@ SELECT *
 7797
 1998-01-10T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -205,12 +185,12 @@ SELECT *
 7797
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-02T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 827
 7797
-1998-01-15T00:00Z[UTC]
+1998-01-20T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
 2020-01-03T00:00Z[UTC]
@@ -238,7 +218,7 @@ SELECT *
 145
 7797
 1998-01-03T00:00Z[UTC]
-1998-01-10T00:00Z[UTC]
+1998-01-05T00:00Z[UTC]
 2020-01-04T00:00Z[UTC]
 2020-01-05T00:00Z[UTC]
 1
@@ -246,40 +226,19 @@ SELECT *
 7797
 1998-01-05T00:00Z[UTC]
 1998-01-10T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
-9999-12-31T23:59:59.999999Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
 2020-01-04T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+9999-12-31T23:59:59.999999Z[UTC]
 1
 145
 7797
 1998-01-10T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -288,12 +247,12 @@ SELECT *
 7797
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-02T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 827
 7797
-1998-01-15T00:00Z[UTC]
+1998-01-20T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
 2020-01-03T00:00Z[UTC]
@@ -324,7 +283,7 @@ SELECT *
 145
 7797
 1998-01-03T00:00Z[UTC]
-1998-01-10T00:00Z[UTC]
+1998-01-05T00:00Z[UTC]
 2020-01-04T00:00Z[UTC]
 2020-01-05T00:00Z[UTC]
 1
@@ -332,7 +291,7 @@ SELECT *
 7797
 1998-01-05T00:00Z[UTC]
 1998-01-10T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-04T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 145
@@ -345,34 +304,20 @@ SELECT *
 145
 7797
 1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+1998-01-12T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
+2020-01-06T00:00Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-12T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -388,12 +333,12 @@ SELECT *
 7797
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-02T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 827
 7797
-1998-01-15T00:00Z[UTC]
+1998-01-20T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
 2020-01-03T00:00Z[UTC]
@@ -417,7 +362,7 @@ SELECT *
 1998-01-05T00:00Z[UTC]
 1998-01-12T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
-NULL
+9999-12-31T23:59:59.999999Z[UTC]
 
 statement ok
 DELETE
@@ -442,7 +387,7 @@ SELECT *
 145
 7797
 1998-01-03T00:00Z[UTC]
-1998-01-10T00:00Z[UTC]
+1998-01-05T00:00Z[UTC]
 2020-01-04T00:00Z[UTC]
 2020-01-05T00:00Z[UTC]
 1
@@ -450,7 +395,7 @@ SELECT *
 7797
 1998-01-05T00:00Z[UTC]
 1998-01-10T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-04T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 145
@@ -463,34 +408,20 @@ SELECT *
 145
 7797
 1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+1998-01-12T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
+2020-01-06T00:00Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-12T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -500,25 +431,18 @@ SELECT *
 1998-01-12T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
-2020-01-07T00:00Z[UTC]
-1
-827
-7797
-1998-01-12T00:00Z[UTC]
-1998-01-20T00:00Z[UTC]
-2020-01-07T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 827
 7797
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-02T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 827
 7797
-1998-01-15T00:00Z[UTC]
+1998-01-20T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
 2020-01-03T00:00Z[UTC]
@@ -532,11 +456,8 @@ SELECT DISTINCT x.xt$system_from
 ----
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
 2020-01-04T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
-2020-01-07T00:00Z[UTC]
 
 query ITT nosort
 SELECT *
@@ -546,6 +467,7 @@ SELECT *
          AS x (customer_number, xt$valid_from, xt$valid_to)
 ----
 
+
 query ITT nosort
 SELECT *
   FROM Prop_Owner
@@ -571,7 +493,6 @@ SELECT *
 1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 
-
 query ITT rowsort
 SELECT *
   FROM Prop_Owner
@@ -589,7 +510,6 @@ SELECT *
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
 
-
 query ITT nosort
 SELECT *
   FROM Prop_Owner
@@ -620,24 +540,9 @@ SELECT *
 ----
 145
 2020-01-01T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-145
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-145
-2020-01-03T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-145
-2020-01-04T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
-145
-2020-01-05T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 827
 2020-01-06T00:00Z[UTC]
-2020-01-07T00:00Z[UTC]
-827
-2020-01-07T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 
 query ITT nosort
@@ -649,7 +554,7 @@ SELECT *
 ----
 827
 2020-01-06T00:00Z[UTC]
-NULL
+9999-12-31T23:59:59.999999Z[UTC]
 
 statement ok
 INSERT INTO Prop_Owner (xt$id, customer_number, property_number, xt$valid_from)
@@ -673,7 +578,7 @@ SELECT *
 145
 7797
 1998-01-03T00:00Z[UTC]
-1998-01-10T00:00Z[UTC]
+1998-01-05T00:00Z[UTC]
 2020-01-04T00:00Z[UTC]
 2020-01-05T00:00Z[UTC]
 1
@@ -681,7 +586,7 @@ SELECT *
 7797
 1998-01-05T00:00Z[UTC]
 1998-01-10T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-04T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 145
@@ -694,34 +599,20 @@ SELECT *
 145
 7797
 1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-02T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-1
-145
-7797
-1998-01-10T00:00Z[UTC]
-1998-01-15T00:00Z[UTC]
-2020-01-04T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+1998-01-12T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
+2020-01-06T00:00Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-12T00:00Z[UTC]
 1998-01-15T00:00Z[UTC]
-2020-01-05T00:00Z[UTC]
+2020-01-01T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 145
 7797
-1998-01-10T00:00Z[UTC]
+1998-01-15T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-01T00:00Z[UTC]
 2020-01-02T00:00Z[UTC]
@@ -731,25 +622,18 @@ SELECT *
 1998-01-12T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
-2020-01-07T00:00Z[UTC]
-1
-827
-7797
-1998-01-12T00:00Z[UTC]
-1998-01-20T00:00Z[UTC]
-2020-01-07T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 1
 827
 7797
 1998-01-15T00:00Z[UTC]
 1998-01-20T00:00Z[UTC]
-2020-01-03T00:00Z[UTC]
+2020-01-02T00:00Z[UTC]
 2020-01-06T00:00Z[UTC]
 1
 827
 7797
-1998-01-15T00:00Z[UTC]
+1998-01-20T00:00Z[UTC]
 9999-12-31T23:59:59.999999Z[UTC]
 2020-01-02T00:00Z[UTC]
 2020-01-03T00:00Z[UTC]
@@ -770,6 +654,7 @@ WHERE P1.property_number = 7797
   AND P1.customer_number = P2.customer_number
 ----
 
+
 query ITT nosort
 SELECT P2.property_number,
        GREATEST(P1.xt$valid_from, P2.xt$valid_from) AS VT_Begin,
@@ -808,6 +693,7 @@ WHERE P1.property_number = 7797
   AND P1.SYSTEM_TIME OVERLAPS P2.SYSTEM_TIME
 ----
 
+
 query ITTTT nosort
 SELECT P2.property_number,
        GREATEST(P1.xt$valid_from, P2.xt$valid_from) AS VT_Begin,
@@ -860,6 +746,7 @@ WHERE P1.property_number = 7797
   AND P1.SYSTEM_TIME CONTAINS PERIOD(P2.xt$system_from, P2.xt$system_to)
 ----
 
+
 query ITTT nosort
 SELECT P2.property_number,
        GREATEST(P1.xt$valid_from, P2.xt$valid_from) AS VT_Begin,
@@ -889,3 +776,4 @@ SELECT P2.property_number, P2.xt$system_from AS Recorded_Start
 ----
 3621
 2020-01-08T00:00Z[UTC]
+