feat(optimize): guarantee global sorting in OPTIMIZE output

[tolleybot]

feat(optimize): guarantee global sorting in OPTIMIZE output

After running OPTIMIZE on a Delta table, files were not guaranteed to be
globally sorted by (objectId, dateTime), forcing downstream consumers to
perform an extra sort. This change adds a global sort phase in the compaction
pipeline and exposes new flags to toggle or customize sorting.

Closes #3538

[github-actions]

ACTION NEEDED

delta-rs follows the Conventional Commits specification for release automation.

The PR title and description are used as the merge commit message. Please update your PR title and description to match the specification.

[ion-elgreco]

Fine by me to add this, I can see the value. Just left a couple comments

[ion-elgreco]

This flag is unnecessary, passing cols should be enough to have it be disabled or not, same applies for the rust builder. Just let some() is enough to trigger a different match arm

[hntd187]

I agree, I think this flag is unnecessary.

[tolleybot]

agreed, I made that change

[ion-elgreco]

Arro3 is required so this try except should exist

[hntd187]

nit also deltalake is spelled incorrectly

[tolleybot]

@ion-elgreco could you clarify your comment concerning the try except. Do you want me to revert it or are you ok with it the way I have it? (besides the spelling)

[hntd187]

I'm not entirely convinced this PR necessary. Why is doing a z-order not appropriate with the optimize? I don't really agree it should be "just done" for the user, even Databricks doesn't automatically order the output, it requires a ZORDER clause in a SQL statement.

[hntd187]

This else branch looks like it's a complicated way to do a z-order as below. I might be misunderstanding, could you explain what is happening here?

[tolleybot]

That else block under if !self.sort_enabled is not doing Z-ordering at all, but I added new global sort path for
compaction:

1. We spin up a DataFusion runtime (with a spillable memory pool).
2. Register the Delta object store under delta-rs://
3. Build a DataFrame over exactly the same files in this merge‐bin via DeltaTableProvider
4. Call df.sort(sort_columns…) on the user-supplied keys
5. Turn the sorted plan back into a Stream
6. Hand that stream into our existing rewrite_files() writer, so all output files are globally ordered.

The actual Z-order logic lives separately in the OptimizeOperations::ZOrder branch (around line 917) where we invoke a custom ZOrderUDF. I’ve expanded the comment above the else-block in optimize.rs to spell this out. Hope that clears it up—please let me know if you’d like any further tweaks!

[hntd187]

I agree, I think this flag is unnecessary.

[hntd187]

nit also deltalake is spelled incorrectly

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 21.03%. Comparing base (2cc8081) to head (cf65fa1).

❗ There is a different number of reports uploaded between BASE (2cc8081) and HEAD (cf65fa1). Click for more details.

HEAD has 4 uploads less than BASE

Flag	BASE (2cc8081)	HEAD (cf65fa1)
	6	2

Additional details and impacted files

@@             Coverage Diff             @@
##             main    #3539       +/-   ##
===========================================
- Coverage   74.23%   21.03%   -53.21%     
===========================================
  Files         150       76       -74     
  Lines       44739    13119    -31620     
  Branches    44739    13119    -31620     
===========================================
- Hits        33213     2759    -30454     
- Misses       9384    10076      +692     
+ Partials     2142      284     -1858     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[hntd187]

So judging by the comment which I might be misunderstanding, this rewrites the entire table?

[tolleybot]

Thanks for flagging that, your reading is correct that the original wording made it sound like we were rewriting the entire table. I’ve updated the comment to make it more clear that we only build a DataFusion plan and rewrite the files in the current compaction batch in sorted order (not the full table)

[hntd187]

You shouldn't need this machinery shouldn't session ctx and proper object stores already be registered?

[tolleybot]

Your correct. The DeltaTableProvider already carries the correct object store, so we don’t need to spin up a custom RuntimeEnv or manually register anything.

[hntd187]

I don't know if this test does what we need here. This shows a single partition is sorted, which is fine, but that's the same as a Z-order. I think you'd need to partition the table and add a bit more data in order to validate it sorts correctly across partitions.

[tolleybot]

Ok, thanks for the feedback. Ill put something together

[tolleybot]

Thanks for catching that. I’ve updated test_optimize_global_sort_enabled so it creates the table partitioned on objectId, now when we call
.optimize().with_sort_columns(&["objectId","dateTime"]), it will scan multiple partitions, globally sort across them, and rewrite each partition in order.
This ensures we’re truly testing cross-partition sorting

[hntd187]

What if I don't want the order to be ascending and nulls first?

[tolleybot]

Ok, Ive updated the code so you can decide if you want ascending or not and how nulls are handled

[hntd187]

Circling back on this, sorry, but it's not clear to me, could you explain what problem you are trying to solve here in more detail? I'm not convinced this PR is necessary and there seem to be a few places where this could produce unintended results and take lots of extra compute time. If you have a specific use case or problem it'd be helpful to know. I ask because your PR seems very tailored towards your data and there maybe a way to accomplish what you are looking for without this PR.

[tolleybot]

@hntd187
What the PR gives you
– A single, atomic compaction + global sort step, all in one transaction.
– Zero extra jobs or manual code: just with_sort_columns(&["key","timestamp"]) on your OptimizeBuilder.
– Only one sort per maintenance window (instead of every read).
– Opt-in flag means no change to anyone else’s Optimize path or performance.

Put another way, everything you’d otherwise have to script yourself gets pulled into the existing Optimize API no extra scans, no schema, and no replace‐where gymnastics.

It could be I am overlooking something. I'm open to suggestions!

[adamreeve]

Circling back on this, sorry, but it's not clear to me, could you explain what problem you are trying to solve here in more detail?

Hi, I work with @tolleybot and can hopefully provide some more context on why we want this feature.

Our use case is that we use Delta Lake to store time-series data, and have consumers that want to read time series and store them in memory with a specific order (by object id then timestamp). We would like to minimise the amount of work these consumers need to do to reorder data.

We can't guarantee a whole table or partition will always be globally ordered like this as data can be split across multiple Parquet files with no inherent order, and any append operations for recent data will break the ordering. But if batches of data that are read from Delta Lake are locally ordered it greatly reduces the work our consumers need to do. Currently we always write ordered data, but the compaction operation arbitrarily reorders data.

For what it's worth, Iceberg allows setting a table order that's used by compaction (https://www.tabular.io/apache-iceberg-cookbook/data-engineering-table-write-order/), so I don't think this requirement is unique to us.

[ohanf]

FWIW I found this issue after reading https://blog.cloudflare.com/logexplorer-ga/ which specifically called out the need for this feature in upstream delta-rs, quoting the relevant section

While the open-source implementation of Delta Lake provides compaction out of the box, we soon encountered an issue when using it for our workloads. Stock compaction merges data files to a desired target size S by sorting the files in reverse order of their size and greedily filling bins of size S with them. By merging logs irrespective of their timestamps, this process distributed ingested batches randomly across merged files, destroying data locality. Despite compaction, a user querying for a specific time frame would still end up fetching hundreds or thousands of files from R2.

For this reason, we wrote a custom compaction algorithm that merges ingested batches in order of their minimum log timestamp, leveraging the min/max statistics mentioned previously. This algorithm reduced the number of overlaps between merged files by two orders of magnitude. As a result, we saw a significant improvement in query performance, with some large queries that had previously taken over a minute completing in just a few seconds.

Bringing it up as another data point for something like this feature being useful

[hntd187]

I appreciate the feedback, but it's still not clear to me what this is trying to solve that z-order does not solve? I noticed in the tests you are using dates, but they are string typed and not date types, are you sure that those types and z-order does not already solve this? I created several test tables with z-order to try they appear to also do what you are trying to do.

[tolleybot]

@hntd187
I agree it can look like a full sort on simple string date tests, because ISO‐formatted dates happen to sort chronologically when treated as strings

To make the difference clearer, I added a Python unit test, "test_python_zorder_vs_global_sort_on_ints", that writes a tiny DataFrame with two integer keys, runs both

dt.optimize.z_order(["objectId","seq"])

and

dt.optimize.compact(sort_columns=["objectId","seq"])

and then compares the outputs:
* The Z-order output does not match the true lexicographic sequence (0,0),(0,1),(0,2),(0,3), proving that interleaving bits reorders rows.
* The global sort output exactly matches the lexicographic sequence, showing that only an explicit sort can guarantee monotonic (objectId, seq) order.

[hntd187]

I checked your test out and slightly modified it to write 2 separate tables. They are physically the same and checksums show the methods produce a byte for byte identical output. I think whatever you are doing to re-read the tables back in is altering the ordering of the rows somehow, but on a physical file level they are identical. Is there something I'm missing here?

[jeromegn]

I'm very interested in this, but as far as I can tell, z ordering should achieve the same?

[tolleybot]

I checked your test out and slightly modified it to write 2 separate tables. They are physically the same and checksums show the methods produce a byte for byte identical output. I think whatever you are doing to re-read the tables back in is altering the ordering of the rows somehow, but on a physical file level they are identical. Is there something I'm missing here?

Can I get a copy of your test code. It would be helpful to try myself

[hntd187]

I checked your test out and slightly modified it to write 2 separate tables. They are physically the same and checksums show the methods produce a byte for byte identical output. I think whatever you are doing to re-read the tables back in is altering the ordering of the rows somehow, but on a physical file level they are identical. Is there something I'm missing here?

Can I get a copy of your test code. It would be helpful to try myself

It's the same as your code, I just write 2 tables out instead of using the same table

[tolleybot]

What this PR adds

1. Atomic compaction + global sort
   A single transaction that both compacts data and enforces a strict (objectId, timestamp, …) ordering—no extra jobs or scripting.
2. Opt-in via the existing Optimize API

   dt.optimize
     .compact(Some(vec!["objectId", "timestamp"]))
     .await?;

Users who don’t pass sort_columns keep today’s behavior and performance.

⸻

Why Z-order ≠ global sort

Technique	How the rows are physically laid out	When it helps most
Z-order (Morton)	Follows a space-filling curve by interleaving the bits of all sort keys. Nearby points in any dimension end up in the same files/row-groups.	Multi-column range scans (e.g., geo or IoT queries that might filter on X or Y or both).
Global (lexicographic) sort	Strict (key1, key2, …) sequence—rows are written exactly as they’ll be read.	Workloads that usually filter hard on the first key and want data materialized in that exact order (e.g., objectId → timestamp time-series analytics).

Concrete example

For two tiny columns (objectId, seq) the layouts diverge:

Row objectId (x) seq (y) Z-order bits → Z Resulting order
A 0 1 00 01 → 1
B 1 0 10 00 → 8
C 0 3 01 01 → 5
D 1 2 11 00 → 12

• Z-order sorts by the Z key → (0, 1), (1, 0), (0, 3), (1, 2)
• Global (lexicographic) sort on (objectId, seq) → (0, 1), (0, 3), (1, 0), (1, 2)

You can view both Z-order and global sort outputs side-by-side in our Python gist, making it easy to compare the differences at a glance.

⸻

Why we want global sort for our time-series tables

• Query pattern: We routinely scan large time ranges for a subset of objects.
• In-memory order matters: Down-stream analytics expect rows already ordered by objectId → timestamp, so a pre-sorted layout avoids an extra Spark or Python sort after reading.
• File-level pruning still works: Because objectId is the first sort key, min/max stats let the engine skip entire files for objects outside the filter set—similar to Z-order, but without the bit-interleaving overhead.
• Maintenance simplicity: Rolling the sort into the existing Optimize.compact() call means one job per maintenance window instead of a separate “sort after compaction” script.

Z-order remains ideal when you need multi-column clustering (e.g., geo or IoT cubes). For our time-series use case, a deterministic, lexicographic sequence on (objectId, timestamp) is both faster to read and simpler to maintain.

[hntd187]

The example above does what you suggest, which makes sense, so I went back and looked at my original test that made identical output and I realized that depending on the type of the associated columns here change the sorting output. If you change the first column to be a string then the output is identical. So sometimes when z-order happens to match up with a global sort sometimes it doesn't.

With this in mind, I'd say that this is very niche and specific to your use case. There is nothing stopping you from employing this strategy yourself, but I think adding this will confuse users on what they need and anyone who doesn't use delta-rs might also be confused by why this is here or might expect it to exist in other places. Additionally, this assumes you use delta-rs and only delta-rs (likely Databricks's own optimize or z-order will wipe this all out) so while I appreciate your effort to contribute this, I think it would be best to decline this PR.

[hntd187]

Closing this, as declined.