[SPARK-18187][SS] CompactibleFileStreamLog should not rely on "compactInterval" to detect a compaction batch

[lw-lin]

What changes were proposed in this pull request?

Problem

Right now CompactibleFileStreamLog uses compactInterval to check if a batch is a compaction batch. However, since this conf is controlled by the user, they may just change it, and CompactibleFileStreamLog will just silently return the wrong answer.

Changes

With this patch the compactInterval configuration should only be used when deciding if we should perform a new compaction. The identification of a compaction vs a delta is done by looking for the compact suffix.

In more details, upon each start/restart, we first examine what metadata files were there, what files among them are compacted files. We introduce knownCompactionBatches to hold the existing compaction batches before this run, and zeroBatch to hold the first batch this run should write to.

Thus we can support the following situations:

  (1) a fresh run
  (2) the previous run with `compactInterval` = 2
      0
  (3) the previous run with `compactInterval` = 2
      0   1.compact
  (4) previous runs with `compactInterval` = 2 and `compactInterval` = 5
      0   1.compact    2   3.compact   4.compact
  (5) previous runs with `compactInterval` = 2 and `compactInterval` = 5
      0   1.compact    2   3.compact   4.compact   5   6   7   8

with:

  (1) `knownCompactionBatches` = (), `zeroBatch` = 0
  (2) `knownCompactionBatches` = (), `zeroBatch` = 1
  (3) `knownCompactionBatches` = (1), `zeroBatch` = 2
  (4) `knownCompactionBatches` = (1, 3, 4), `zeroBatch` = 5
  (5) `knownCompactionBatches` = (1, 3, 4), `zeroBatch` = 9

Note this appoach does not require the compactInteral for a new run must be consistent with previous runs.

How was this patch tested?

added test cases

[uncleGen]

related to #15827

[SparkQA]

Test build #68402 has finished for PR 15828 at commit 72de68d.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class CompactibleFileStreamLogSuite extends SparkFunSuite with SharedSQLContext
  • class FakeCompactibleFileStreamLog(

[SparkQA]

Test build #68412 has finished for PR 15828 at commit 58c1031.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #68454 has finished for PR 15828 at commit 7126665.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[lw-lin]

rebased.

• the first & second commits were a simple refactor of the CompactibleFileStreamLogSuite.scala introduced in [SPARK-18425][Structured Streaming][Tests] Test CompactibleFileStreamLog directly #15870
• the third commit was the actual change of this patch

[lw-lin]

Jenkins retest this please

[SparkQA]

Test build #68584 has finished for PR 15828 at commit ca5e984.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #68585 has finished for PR 15828 at commit ca5e984.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.