Add support for writing log compaction files by nicklan · Pull Request #4334 · delta-io/delta

nicklan · 2025-03-27T18:41:51Z

Which Delta project/connector is this regarding?

Description

This adds support for writing log compaction files as a post-commit hook.

This introduces a new method on TransactionBuilder: withLogCompactionInverval, which allows saying how many commits there should be in between log compactions. This is set to 10 by default, and can be set to 0 to disable compaction.

In this PR we count commits just by the commit number. This means checkpoints could cause a compaction to happen more frequently that exactly X json commit files, but this won't harm anything, it's just a slightly more frequent compaction that strictly necessary.

How was this patch tested?

Added unit tests that the actions are as expected in the compacted file, as well as a test that delta-spark can read both with and without the compacted file and get the same result.

Does this PR introduce any user-facing changes?

Yes, this means your log will compact by default, and there is a new method to call on TransactionBuilder if you want to configure it.

scottsand-db

Looks great! Left some minor comments.

allisonport-db · 2025-04-05T00:23:26Z

    return this;
  }

+  @Override


This feels like a really weird config to be on the transaction builder since it doesn't apply to every transaction. Is there really no table property for this?

Maybe this is another scenario where it would make sense for us to have sort of session config.. (and we're kind of bypassing it by putting it here)

Also slightly confusing semantics about when we compact since it's writer based (and not required), but there's probably nothing we can do about that. But it's kind of confusing, do we compact when it's been logCompactionInterval commits since the last compaction? Since the last time this writer compacted based on this writer's number of commits? Based on mod (how it's implemented)?

Not sure if it's worth clarifying this to the user

It probably should/could be an actual config on the table, but that would be a bigger change. Do we have any notion of a "session config"? (afaict, we don't).

Unless there's a current place it fits better I'd advocate for having it here and then we can move it if we introduce something better?

I think we've just been putting off adding something like a session config for a while. I guess this is another example of where it would make sense to have one

Makes sense. Are you okay with that being a follow-up and having this here?

My 2c is: yes that should be a followup. That's quite a large change (designing a configuration API and then plumbing it through everywhere). All of these APIs are maked as @Evolving -- so, IMO, it's fine to add this for now and refactor later 👍 Also curious what Alli things here, too.

+1 I am ok following up later on this. It is a fairly large task to design configuration.

I'm okay with this being a follow up task -- just wanted to note this is another scenario where it seems like we should have a configuration (and curious if others agree)

allisonport-db · 2025-04-05T00:32:26Z

+  }
+
+  /** Utility to convert an Iterator<FilteredColumnarBatch> into an Iterator<Row> */
+  private static class FilteredBatchToRowIter implements CloseableIterator<Row> {


I think this could just be distilled to a flatMap right? @vkorukanti do you know if we ever added this utility somewhere I think we've discussed it a few times

It totally can! But we don't have flatMap available afaict? Java has it for Stream but our iterators don't implement that.

We would implement our own version of it on CloseableIterator (our own interface anyways) or in a utilities class but I thought maybe we had done this? If not maybe we can add this as a util here? I know we've wanted this before

Okay. I'm checking with @vkorukanti if we already have this, otherwise I can add it.

We don't have one. Planning to add one.

However in my opinion we should change or add our JsonHandler to also take the ColumnarBatch and create json rows out of the batches. Delta sharing found that instead of getting rows if they used column vectors in batch directly, it was significant (~40% better) perf improvement.

Thanks Venki!

Sounds like this is still a discussion point. @allisonport-db thoughts on just having this utility here and then moving it to a more general spot if/when we settle on exactly what we want to do?

If it's not too much work to just add it as a util I think I'd prefer that so we can re-use it in the future the next time we need it. Otherwise each time we just keep re-implementing the same general functionality and the duplication grows linearly... But if you don't want to do it in this PR that's okay with me

Plus maybe better since I feel like there's a small risk we make little mistakes each time we rewrite our nested iterators (& maybe don't test all the edge cases)

Sounds good, move to Utils

vkorukanti

LGTM, thanks for adding this feature to Kernel.

vkorukanti · 2025-04-11T05:31:39Z

    return this;
  }

+  @Override


+1 I am ok following up later on this. It is a fairly large task to design configuration.

vkorukanti · 2025-04-11T05:38:37Z

+                Optional.of(endVersion),
+                false /* mustBeRecreatable */)
+            .toInMemoryList();
+


assert to deltas.size == (endVersion - startVersion)?

Isn't it that we could race with a checkpoint, in which case we might get a smaller set?

Actually, I just added this because I think it makes sense to validate and get an error if something unexpected happens.

vkorukanti · 2025-04-11T05:40:50Z

+
+    LogSegment segment =
+        new LogSegment(dataPath, endVersion, deltas, emptyList(), lastCommitTimestamp);
+    CreateCheckpointIterator checkpointIterator =


one nit: should we rename CreateCheckpointIterator to LogCompactionIterator to be generic?

I think this is okay. Log compaction is a "minor checkpoint" so it's not too confusing

vkorukanti · 2025-04-11T05:57:03Z

+      val batch = actions.next().getColumnarBatch()
+      val rows = batch.getRows()
+      while (rows.hasNext()) {


there are some implicit methods added on CloseableIterator in test (scala code - TestUtils). You can just say actions.toSeq.map etc.

allisonport-db · 2025-04-11T16:40:22Z

+        dataPath,
+        System.currentTimeMillis() - startTimeMillis);
+
+    if (deltas.isEmpty()) {


For log segments we do some validation on the deltas found - do you think we should do this here? We can probably reuse the same code

I think it just checks that endVersion and startVersion are present and the deltas are contiguous

Oh it seems like we do this in the LogSegment constructor :) awesome! nevermind here

I think +1 on @vkorukanti comment however, it seems like we check that the endVersion is present and that they are contiguous, but not the first version

Also this can be a follow-up but maybe we should consider the desired behavior in this scenario when files are missing or something -- should we throw an error? and if so what kind? (currently will throw illegal argument exceptions but in some cases it seems like it should be InvalidTableException)

As above, i think a checkpoint racing could cause an issue here. Maybe we want an error in that case though?

allisonport-db · 2025-04-11T17:24:00Z

+          // add
+          val addRow = row.getStruct(1)
+          val path = addRow.getString(0)
+          if (!removed.contains(path)) {


Is this an issue if it doesn't include the key for DVs?

In general yes, but not here as we know what's happening

allisonport-db · 2025-04-11T17:25:39Z

+    var seenMetadata = false
+    var seenProtocol = false
+    val resBuilder = Seq.newBuilder[TestRow]
+    while (actions.hasNext()) {


I feel like essentially rewriting log replay here in order to test this is kind of weird. Wondering if instead we can just do a super simple example and actually check for the adds/removes/metadatas that we know we committed and should be there? open to thoughts

i.e.

version 0 - metadata1 version 1 - add1, add2 version 2 - remove1, add3, dm1 version 3 - dm1_2, metadata2 // etc expect metadata2, dm1_2, add3, remove1, add2

I felt the same. I think self-contained tests where the expected output is hardcoded will make it easier to understand the tests.

allisonport-db · 2025-04-11T17:27:13Z

+        } else if (!rowIsNull(row)) {
+          resBuilder += TestRow(row)


what do we do with domain metadatas here? don't those also have to be resolved with themselves? (only include the latest one)

Great catch, we need to resolve those too. I've added that into the test now

allisonport-db · 2025-04-11T17:34:20Z

+    val testMsgUpdate = if (includeRemoves) " and removes" else ""
+    test(s"Read table with adds$testMsgUpdate") {
+      withTempDirAndEngine { (tablePath, engine) =>
+        addData(tablePath, alternateBetweenAddsAndRemoves = includeRemoves, numberIter = 10)


are we sure we haven't done any compaction by default here? should we disable it explicitly?

oss spark doesn't support writing minor compactions

Ah did not realize this is done using Spark. Anyway to make sure if that is added to Spark this test doesn't start writing them?

Yeah, I mean if it's added to spark we'd want to disable compaction, but we can't preemptively do that since there's no config to set currently.

nicklan · 2025-04-11T22:33:18Z

+    val seenDomains = scala.collection.mutable.HashSet.empty[String]
+    var seenMetadata = false
+    var seenProtocol = false
+    actions.toSeq.flatMap { wrapper =>


resurrecting #4334 (comment)

Talked a bit with venki. There's two main issues to just statically declaring expected results:

The file names in the add/remove actions are different each time. So we'd have to somehow discover those

Constructing a TestRow with a wide schema with nested structs is pretty messy. You need to do something like TestRow(null, null, addFile(....), null, null, null, null), and I'm still not quite clear what addFile would have to return to make it work.

Really we just want log replay here, but it would be obviously better for us to not re-write it for the test.

I couldn't find an easy way to expose kernel log replay to get just the reconciled actions (other than the checkpoint iterator which is actually what we're trying to test here). Perhaps there's a way to get oss-spark to do that, but I haven't found that yet either.

@allisonport-db would you be okay with taking that as a follow-up since I think this is testing that the feature works, we could just probably clean it up and remove a bunch of code if we figure out the above.

Fine for a follow up. I think we're picturing different tests though I was thinking
(1) manually commit add/removes so we know the paths
(2) we don't need to check the exact contents necessarily in this test, but assert for example add(path=/something/file1) exists and maybe add(path=/something/file2) does not exist etc
Or maybe a different way to do it using the actual adds we commit to compare

Future ideas for follow up too is we can use the spark API to correctly check the DM as well I think, also would be good to have a test where theres a metadata + protocol update in the version range (make sure we keep the latest one only)

We do have that test btw. The domain metadata changed enables the feature so it writes new p and m

allisonport-db · 2025-04-14T20:05:53Z

+        hook.threadSafeInvoke(engine)
+
+        spark.conf.set(DeltaSQLConf.DELTALOG_MINOR_COMPACTION_USE_FOR_READS.key, "true")
+        val withCompactionData = readUsingSpark(tablePath)


Only an idea but if you include domain metadata you can check those with spark as well by loading the snapshot and accessing snapshot.domainMetadata

DeltaLog.forTable(spark, tablePath).getSnapshotAt(version) to get the Spark snapshot in Kernel btw

allisonport-db

LGTM but let's revisit the testing as a followup (I feel like there's a few different options we could consider)

allisonport-db · 2025-04-15T02:50:48Z

  }
+
+  // Utility class to support `intoRows` below
+  private static class FilteredBatchToRowIter implements CloseableIterator<Row> {


Sorry I think maybe I wasn't clear; I meant we should add a flatMap util to the CloseableIterator interface and we can generally use that for everything else. Fine to do this in a follow-up though don't want to block on this

nicklan · 2025-04-15T23:25:45Z

@allisonport-db I've added #4416 and #4417 as follow-ups. I think that should make us good to merge. LMK if you need anything else. Thanks!

nicklan added 4 commits March 26, 2025 16:15

initial with outline of code

80f972d

fix test

3ca599c

get minFileRetentionTimestampMillis from metadata

fd3645d

remove "minor" from everything

fa8ebda

nicklan requested review from allisonport-db, raveeram-db, scottsand-db and vkorukanti and removed request for allisonport-db, scottsand-db and vkorukanti March 27, 2025 18:41

nicklan changed the title ~~Add support for log compaction~~ Add support for writing log compaction files Mar 27, 2025

one final rename

7f770ca

nicklan self-assigned this Mar 27, 2025

scottsand-db reviewed Mar 27, 2025

View reviewed changes

allisonport-db reviewed Apr 5, 2025

View reviewed changes

nicklan added 6 commits April 7, 2025 16:23

address comments

f4edf6a

add test

b2c8739

Add initial tests

fad8e30

add test where we use spark to read with/without compaction

9e895ea

include testing DM and PandM dedup

fce9ffd

add comment

fec72c7

nicklan requested review from allisonport-db and scottsand-db April 10, 2025 00:26

nicklan marked this pull request as ready for review April 10, 2025 00:27

nicklan added 2 commits April 9, 2025 17:35

Merge branch 'master' into minor-log-compaction

d00bd57

fmt

2a8a61c

nicklan added 4 commits April 9, 2025 18:11

add test that hooks are created

0ba6efb

lots of small fixups to generate correctly

59d6bf8

use asScala

69592c3

allow hella args

19bb58e

vkorukanti approved these changes Apr 11, 2025

View reviewed changes

allisonport-db reviewed Apr 11, 2025

View reviewed changes

nicklan added 2 commits April 11, 2025 14:36

address some comments

0078d08

address more comments

cfd832d

nicklan commented Apr 11, 2025

View reviewed changes

nicklan added 2 commits April 11, 2025 15:38

moar flatmap

8aad859

move iterator def to utils

5d61674

allisonport-db reviewed Apr 14, 2025

View reviewed changes

validate number of delta files found

19c045a

nicklan requested a review from allisonport-db April 15, 2025 00:24

allisonport-db approved these changes Apr 15, 2025

View reviewed changes

Merge remote-tracking branch 'upstream/master' into minor-log-compaction

f7c357a

nicklan mentioned this pull request Apr 15, 2025

[Feature Request] Add a flatMap util to the CloseableIterator #4416

Open

8 tasks

fix comment

b4f27ce

allisonport-db merged commit 9d3a8e7 into delta-io:master Apr 16, 2025

Uh oh!

Conversation

nicklan commented Mar 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which Delta project/connector is this regarding?

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

Uh oh!

scottsand-db left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vkorukanti left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

nicklan commented Mar 27, 2025 •

edited

Loading

vkorukanti Apr 11, 2025 •

edited

Loading