[Kernel] Minor refactor to DeltaLogActionUtils; add CloseableIterator takeWhile and other helpful methods

[scottsand-db]

Which Delta project/connector is this regarding?

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

Description

This PR does the following:

• changes our DeltaLogActionUtils listDeltaLogFiles method to return an iterator. This will eventually let it be used in a followup PR to further consolidate and clean up code in DeltaHistoryManager (which lists the delta log and expects an iterator)
• adds CloseabelIterator::takeWhile, breakableFilter, and toInMemoryList

How was this patch tested?

• New UTs

Does this PR introduce any user-facing changes?

No.

[scottsand-db]

verbatim translate this logic to use the breakableFilter API instead

[huan233usc]

LGTM with the under radar doc nit got fix(https://github.com/delta-io/delta/pull/4097/files#r1933026458)

[allisonport-db]

just 1 tiny comment + a question

[allisonport-db]

Is this a common paradigm? It seems like breakableFilter could also be a combo of filter + takeWhile. But I see those are implemented using this so just curious on your reasoning

[scottsand-db]

breakableFilter could also be a combo of filter + takeWhile

Are you saying that we can implement breakableFilter by using a combination of both filter + takeWhile?

One of my tests has an example of breakableFilter: we want to: INCLUDE 1, EXCLUDE 2, INCLUDE 3, and BREAK at 4. After thinking about this for > 30 seconds, maybe there is a way to implement this using filter + takeWhile, but it doesn't jump out at me.

I think the semantics I lay out here are simpler than using .filter(....).takeWhile(...). For example, the implementation I did in DeltaLogActionUtils is cleanly implemented this way.

[allisonport-db]

    val result = throwingDataIter.breakableFilter { x =>
      if (x <= 1 || x == 3) {
        BreakableFilterResult.INCLUDE
      } else if (x == 2) {
        BreakableFilterResult.EXCLUDE
      } else if (x == 4) {
        BreakableFilterResult.BREAK
      } else {
        throw new RuntimeException("This should never be reached")
      }
    }

Wouldn't the test case just be? They seem equivalent to me

iter.filter( x => x != 2).takeWhile(x != 4)

[allisonport-db]

Equivalently the snapshot manager code would probably be

    // Must be final to be used in lambda
    final AtomicBoolean hasReturnedAnElement = new AtomicBoolean(false);

    return listLogDir(engine, tablePath, startVersion)
        .filter(
            fs -> {
              if (fileTypes.contains(DeltaLogFileType.COMMIT)
                  && FileNames.isCommitFile(getName(fs.getPath()))) {
                return true
              } else if (fileTypes.contains(DeltaLogFileType.CHECKPOINT)
                  && FileNames.isCheckpointFile(getName(fs.getPath()))
                  && fs.getSize() > 0) {
                return true
              } else {
                return false
              }
           }
      .takeWhile(
          fs -> {
              final long fileVersion = FileNames.getFileVersion(new Path(fs.getPath()));

              if (fileVersion < startVersion) {
                throw new RuntimeException(
                    String.format(
                        "Listing files in %s with startVersion %s yet found file %s.",
                        logPath, startVersion, fs.getPath()));
              }

              if (endVersionOpt.isPresent()) {
                final long endVersion = endVersionOpt.get();

                if (fileVersion > endVersion) {
                  if (mustBeRecreatable && !hasReturnedAnElement.get()) {
                    final long earliestVersion =
                        DeltaHistoryManager.getEarliestRecreatableCommit(engine, logPath);
                    throw DeltaErrors.versionBeforeFirstAvailableCommit(
                        tablePath.toString(), endVersion, earliestVersion);
                  } else {
                    logger.debug(
                        "Stopping listing; found file {} with version greater than endVersion {}",
                        fs.getPath(),
                        endVersion);
                    return false;
                  }
                }
              }

              hasReturnedAnElement.set(true);

              return true;
            });

[allisonport-db]

I actually kind of like it; since it nicely delineates the filter mechanism versus the terminate the loop (takeWhile) mechanism. But they honestly are super similar so I don't feel too strongly about it.

[scottsand-db]

iter.filter( x => x != 2).takeWhile(x != 4) --> ah, I understand now, thanks for showing that.

I actually kind of like it; since it nicely delineates the filter mechanism versus the terminate the loop (takeWhile) mechanism.

I actually still prefer the current version. When they are separate, I need to understand 4 return values: 2 for the filter, and 2 for the takeWhile. It also introduces some coupling -- the takeWhile logic is inherintly tied to the filter logic -- e.g. it would we weird to filter(x -> x < 10) and then takeWhile(x > 20). IMO when it's all together, I think it's easier to parse the logic.

[scottsand-db]

The good news is that this CloseableIterator API doesn't preclude your suggestion. i.e. other areas in the code are free to do .filter and then .takeWhile if we decide later on that that is best there.

Overall -- I'm inclined to stick with what I have now. Happy to refactor it laterr -- but really I'd like to unblock @huan233usc 's PR.

[allisonport-db]

Sounds good

[allisonport-db]

I guess I just hesitate to add a new concept of this breakable filter when it can be distilled into two existing things. But I agree it's easy to understand so not a big deal