[Spark] Fix metadata cleanup by retaining a checkpoint before the cutoff window. Alternative fix

[felipepessoto]

Which Delta project/connector is this regarding?

• Spark
• Standalone
• Flink
• Kernel
• Other (fill in here)

Description

Delete eligible delta log files only if there's a checkpoint newer than them before the cutoff window.

Resolves #606

Unit tests based on #2673

How was this patch tested?

Unit Tests

Does this PR introduce any user-facing changes?

Yes, tables with low rate of commit/checkpoints will have increased log retention beyond the cutoff window

[felipepessoto]

@andreaschat-db, could you help with the Metadata cleanup respects requireCheckpointProtectionBeforeVersion test failure, please?

I couldn't find any info in Delta protocol about checkpointProtection table feature

[info] - Metadata cleanup respects requireCheckpointProtectionBeforeVersion *** FAILED ***
[info]   HashSet(10, 14, 6, 9, 13, 12, 7, 11, 8, 15) did not equal HashSet(10, 14, 9, 13, 12, 11, 8, 15) (CheckpointProtectionTestUtilsMixin.scala:91)
[info]   Analysis:
[info]   HashSet(missingInRight: [6, 7])
[info]   org.scalatest.exceptions.TestFailedException:
[info]   at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
[info]   at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
[info]   at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
[info]   at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.$anonfun$testRequireCheckpointProtectionBeforeVersion$1(CheckpointProtectionTestUtilsMixin.scala:91)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.$anonfun$testRequireCheckpointProtectionBeforeVersion$1$adapted(CheckpointProtectionTestUtilsMixin.scala:38)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir(DeltaSQLTestUtils.scala:52)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir$(DeltaSQLTestUtils.scala:50)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.withTempDir(DeltaRetentionSuite.scala:38)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir(DeltaSQLTestUtils.scala:35)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir$(DeltaSQLTestUtils.scala:34)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.withTempDir(DeltaRetentionSuite.scala:38)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.testRequireCheckpointProtectionBeforeVersion(CheckpointProtectionTestUtilsMixin.scala:38)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.testRequireCheckpointProtectionBeforeVersion$(CheckpointProtectionTestUtilsMixin.scala:30)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.testRequireCheckpointProtectionBeforeVersion(DeltaRetentionSuite.scala:38)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.$anonfun$new$94(DeltaRetentionSuite.scala:819)

[andreaschat-db]

@andreaschat-db, could you help with the Metadata cleanup respects requireCheckpointProtectionBeforeVersion test failure, please?

I couldn't find any info in Delta protocol about checkpointProtection table feature

[info] - Metadata cleanup respects requireCheckpointProtectionBeforeVersion *** FAILED ***
[info]   HashSet(10, 14, 6, 9, 13, 12, 7, 11, 8, 15) did not equal HashSet(10, 14, 9, 13, 12, 11, 8, 15) (CheckpointProtectionTestUtilsMixin.scala:91)
[info]   Analysis:
[info]   HashSet(missingInRight: [6, 7])
[info]   org.scalatest.exceptions.TestFailedException:
[info]   at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
[info]   at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
[info]   at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
[info]   at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.$anonfun$testRequireCheckpointProtectionBeforeVersion$1(CheckpointProtectionTestUtilsMixin.scala:91)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.$anonfun$testRequireCheckpointProtectionBeforeVersion$1$adapted(CheckpointProtectionTestUtilsMixin.scala:38)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir(DeltaSQLTestUtils.scala:52)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir$(DeltaSQLTestUtils.scala:50)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.withTempDir(DeltaRetentionSuite.scala:38)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir(DeltaSQLTestUtils.scala:35)
[info]   at org.apache.spark.sql.delta.test.DeltaSQLTestUtils.withTempDir$(DeltaSQLTestUtils.scala:34)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.withTempDir(DeltaRetentionSuite.scala:38)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.testRequireCheckpointProtectionBeforeVersion(CheckpointProtectionTestUtilsMixin.scala:38)
[info]   at org.apache.spark.sql.delta.CheckpointProtectionTestUtilsMixin.testRequireCheckpointProtectionBeforeVersion$(CheckpointProtectionTestUtilsMixin.scala:30)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.testRequireCheckpointProtectionBeforeVersion(DeltaRetentionSuite.scala:38)
[info]   at org.apache.spark.sql.delta.DeltaRetentionSuite.$anonfun$new$94(DeltaRetentionSuite.scala:819)

Hi @felipepessoto, please check the doc in TableFeature. AFAIU, your PR is changing which checkpoints are removed and this changes the expectations of the test.