Add Delta Lake version history to IndexLogEntry for efficient time travel query #272

sezruby · 2020-11-25T02:32:54Z

What is the context for this pull request?

Tracking Issue: Improvement of Time Travel Query on Delta Lake #270
Parent Issue: Hyperspace for Delta Lake #148
Dependencies: Support Delta Lake file-based source provider #265 (merged)

Fixes #270
`

What changes were proposed in this pull request?

This PR improves Hybrid Scan for a time travel query of Delta Lake table by using an old version of index which is closest to the given time travel delta version.

This PR includes:

keep Delta Lake table versions at each create/refresh indexes in IndexLogEntry metadata.
- in CoveringIndex.properties.properties, (to avoid searching all index log entry files)
  - key: DELTA_VERSION_HISTORY_PROPERTY
  - value: comma separated "INDEX_VERSION_ID:DELTA_VERSION_ID"
    - e.g 1:1,3:5,5:7,7:10,9:15 for refreshed delta version 1, 5, 7, 10, 15
find and replace an old version of index in getCandidateIndexes
- new closestIndexVersion API in source provider

Does this PR introduce any user-facing change?

Yes, an old version of index can be applied for a time travel query on Delta Lake.

How was this patch tested?

Unit tests

src/main/scala/com/microsoft/hyperspace/actions/CreateActionBase.scala

sezruby · 2020-11-25T02:54:59Z

src/main/scala/com/microsoft/hyperspace/actions/CreateAction.scala

    with Action {
-  final override def logEntry: LogEntry = getIndexLogEntry(spark, df, indexConfig, indexDataPath)
+  final override def logEntry: LogEntry =
+    getIndexLogEntry(spark, df, indexConfig, indexDataPath, super[Action].endId)


Note: need to get the index version id, but it's not available in CreateActionBase.getIndexLogEnrtry.
So I added endId function and with Action

src/main/scala/com/microsoft/hyperspace/index/rules/RuleUtils.scala

sezruby · 2020-12-10T02:43:14Z

@pirz @apoorvedave1 Could you review the change and give any comment on the approach - keeping version history in properties -? Thanks!

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeFileBasedSource.scala

thugsatbay · 2021-01-28T21:16:32Z

src/test/scala/com/microsoft/hyperspace/index/DeltaLakeIntegrationTest.scala

+          // Get the timestamp string for time travel query using "timestampAsOf" option.
+          val timestampStr = getSparkFormattedTimestamps(System.currentTimeMillis).head
+          // Sleep 1 second because the unit of commit timestamp is second.
+          Thread.sleep(1000)


Avoid using sleep in ut as it will take too much time especially when this method is called multiple times.

To simulate time just take a global Time variable currTime = System.currentTimeMillis and then keep on adding 1000 to it whenever you want to move time and then just use with getSparkFormattedTimestamps function

I added this because delta lake internally records the time in second unit. So I forces to record different times for each version update, so that I could test the query with timestamp properly.

I don't think so, I ran your test and used DeltaTable.forPath(spark, dataPath).history().show(10, false) to view timestamp (2021-02-02 14:49:19.077). It used milliseconds and each op should have a readVersion ?

You can use the timestamp in delta table to store read versions ?

Seems Azure test framework needs sleep for some reason :D.. f19614b
I might have added this after debugging / investigation.

I added this because delta lake internally records the time in second unit. So I forces to record different times for each version update, so that I could test the query with timestamp properly.

Maybe add this comment to the code? Reading the existing comment, I didn't know why we had to sleep.

thugsatbay · 2021-01-28T21:27:59Z

src/main/scala/com/microsoft/hyperspace/actions/Action.scala

 */
 trait Action extends HyperspaceEventLogging with Logging with ActiveSparkSession {
  protected val baseId: Int = logManager.getLatestId().getOrElse(-1)
+  protected def endId: Int = baseId + 2


Sorry, why +2, can you please explain ?

it's because we write the log twice at begin() using baseId + 1 and end() using baseId + 2.
As begin() has in-progress state (e.g. creating), Hyperspace always reads the log file written in "end()"

nit: Can we rename it to firstLogEntryId and lastLogEntryId. And add a comment for the same.
Also should we have an assert that firstLogEntryId would always be -1 or an odd number ?

baseId is not firstLogEntryId and it requires some change..
If an operation failed in op(), then firstLogEntryId could be an odd number. (though failures are not handled properly now. e.g. #248)

cc @apoorvedave1

this is required because LogEntry.id is not available when creating - IndexLogEntry

Can we add final modifier?

imback82 · 2021-02-02T06:59:12Z

@sezruby Could you fix the conflicts?

Unit tests (test will be added)

This can be removed now?

jaceklaskowski · 2021-02-02T09:33:33Z

src/main/scala/com/microsoft/hyperspace/index/sources/FileBasedSourceProviderManager.scala

+  }
+
+  /**
+   * Returns index version related properties.


nit: That looks like something for @return below?

Yes, but this line is consistent with src/main/scala/com/microsoft/hyperspace/index/sources/interfaces.scala and other functions/source providers also have the comments in the same way.
Let's keep this for consistency for now. Thanks!

src/main/scala/com/microsoft/hyperspace/index/sources/default/DefaultFileBasedSource.scala

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeFileBasedSource.scala

imback82 · 2021-02-12T21:50:45Z

@sezruby Could you resolve the conflicts? Let's try to get this in next (sorry for the delay).

@apoorvedave1 Can you also help with the review when you get a chance? Thanks.

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeRelation.scala

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala

…timetravel

src/main/scala/com/microsoft/hyperspace/index/sources/interfaces.scala

src/main/scala/com/microsoft/hyperspace/actions/CreateActionBase.scala

src/main/scala/com/microsoft/hyperspace/actions/RefreshAction.scala

src/main/scala/com/microsoft/hyperspace/actions/RefreshIncrementalAction.scala

src/main/scala/com/microsoft/hyperspace/index/IndexConstants.scala

src/main/scala/com/microsoft/hyperspace/index/IndexCollectionManager.scala

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeRelation.scala

…timetravel

imback82

Almost there

src/main/scala/com/microsoft/hyperspace/actions/OptimizeAction.scala

src/main/scala/com/microsoft/hyperspace/actions/RefreshActionBase.scala

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala

imback82 · 2021-03-21T20:02:16Z

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala

+   * @param logVersion Index log version to retrieve.
+   * @return IndexLogEntry if the index of the given log version exists, otherwise None.
+   */
+  def getIndexLogEntry(


We have def getIndexes(states: Seq[String] = Seq()): Seq[IndexLogEntry]. Should we make the naming consistent? Also, we should expose an API to return log versions: #272 (comment).

If you want, you can create a separate PR that just updates IndexManager APIs, but I am also fine if you do it in this PR.

will do this with a follow up PR

sounds good!

src/main/scala/com/microsoft/hyperspace/index/sources/interfaces.scala

src/test/scala/com/microsoft/hyperspace/index/DeltaLakeIntegrationTest.scala

imback82 · 2021-03-21T20:43:33Z

src/test/scala/com/microsoft/hyperspace/index/DeltaLakeIntegrationTest.scala

+          // Get the timestamp string for time travel query using "timestampAsOf" option.
+          val timestampStr = getSparkFormattedTimestamps(System.currentTimeMillis).head
+          // Sleep 1 second because the unit of commit timestamp is second.
+          Thread.sleep(1000)


I added this because delta lake internally records the time in second unit. So I forces to record different times for each version update, so that I could test the query with timestamp properly.

Maybe add this comment to the code? Reading the existing comment, I didn't know why we had to sleep.

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeRelation.scala

src/test/scala/com/microsoft/hyperspace/index/DeltaLakeIntegrationTest.scala

imback82 · 2021-03-22T01:04:56Z

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala

+   * @param logVersion Index log version to retrieve.
+   * @return IndexLogEntry if the index of the given log version exists, otherwise None.
+   */
+  def getIndexLogEntry(


sounds good!

imback82

LGTM (few minor comments), thanks @sezruby!

imback82 · 2021-03-22T01:11:16Z

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeRelation.scala

+        .getOrElse(DeltaLakeConstants.DELTA_VERSION_HISTORY_PROPERTY, "")
+
+    // The value is comma separated versions - <index log version>:<delta table version>.
+    // e.g. "1:2,3:5,5:9"


// Versions are processed in a reverse order to keep the higher index log version in case different index // log versions refer to the same delta lake version. // For example, "1:1, 2:2, 3:2" will become Seq((1, 1), (3, 2)).

, if I understand this correctly?

sezruby commented Nov 25, 2020

View reviewed changes

sezruby changed the title ~~[WIP] Add Delta Lake version history for efficient Hybrid Scan~~ [WIP] Add Delta Lake version history for efficient time travel query Nov 25, 2020

sezruby commented Nov 25, 2020

View reviewed changes

src/main/scala/com/microsoft/hyperspace/index/rules/RuleUtils.scala Outdated Show resolved Hide resolved

sezruby force-pushed the source_extension_timetravel branch from d843425 to 6e8b2d6 Compare November 25, 2020 09:12

sezruby self-assigned this Nov 25, 2020

sezruby added this to the November 2020 milestone Nov 25, 2020

sezruby added advanced issue This is the tag for advanced issues which involve major design changes or introduction enhancement New feature or request labels Nov 25, 2020

sezruby requested review from apoorvedave1, imback82 and pirz November 25, 2020 09:15

sezruby changed the title ~~[WIP] Add Delta Lake version history for efficient time travel query~~ Add Delta Lake version history to IndexLogEntry for efficient time travel query Nov 25, 2020

sezruby force-pushed the source_extension_timetravel branch 2 times, most recently from 3dc848c to 66e736f Compare November 26, 2020 10:47

sezruby force-pushed the source_extension_timetravel branch from 66e736f to 862bb98 Compare December 9, 2020 06:42

sezruby mentioned this pull request Dec 22, 2020

Hyperspace for Delta Lake #148

Open

4 tasks

sezruby force-pushed the source_extension_timetravel branch from 4534867 to 8a3929c Compare December 22, 2020 08:02

sezruby force-pushed the source_extension_timetravel branch from e9a9b8a to 7f3fc2c Compare January 14, 2021 08:22

This was referenced Jan 22, 2021

Introduce IndexHadoopFsRelation to show applied index name & version in query plan #323

Merged

Fix wrong use of map of Set for summation #336

Merged

sezruby force-pushed the source_extension_timetravel branch from 1bd3a77 to c415baf Compare January 27, 2021 09:33

thugsatbay reviewed Jan 28, 2021

View reviewed changes

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeFileBasedSource.scala Outdated Show resolved Hide resolved

thugsatbay reviewed Jan 28, 2021

View reviewed changes

imback82 modified the milestones: November 2020, February 2021 (v0.5.0) Feb 2, 2021

jaceklaskowski reviewed Feb 2, 2021

View reviewed changes

imback82 reviewed Feb 20, 2021

View reviewed changes

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeRelation.scala Outdated Show resolved Hide resolved

imback82 reviewed Feb 20, 2021

View reviewed changes

src/main/scala/com/microsoft/hyperspace/index/sources/delta/DeltaLakeRelation.scala Show resolved Hide resolved

imback82 reviewed Feb 20, 2021

View reviewed changes

src/main/scala/com/microsoft/hyperspace/index/IndexManager.scala Outdated Show resolved Hide resolved

review commit

7615f92

sezruby force-pushed the source_extension_timetravel branch from ded45b6 to 7615f92 Compare March 2, 2021 03:33

sezruby added 3 commits March 1, 2021 19:46

Merge remote-tracking branch 'upstream/master' into source_extension_…

1469cdb

…timetravel

Build fix

af1a2b3

Merge remote-tracking branch 'upstream/master' into source_extension_…

936c062

…timetravel

imback82 reviewed Mar 11, 2021

View reviewed changes

src/main/scala/com/microsoft/hyperspace/index/sources/interfaces.scala Outdated Show resolved Hide resolved

src/main/scala/com/microsoft/hyperspace/index/sources/interfaces.scala Outdated Show resolved Hide resolved

review commit4

93bd03c

sezruby requested review from imback82 and removed request for pirz March 19, 2021 02:09

imback82 reviewed Mar 21, 2021

View reviewed changes

sezruby mentioned this pull request Mar 21, 2021

Check removed index log versions of time travel query index candidates #387

Closed

sezruby added 3 commits March 20, 2021 22:34

review commit 222

a0f690b

Merge remote-tracking branch 'upstream/master' into source_extension_…

366d999

…timetravel

fix test

647bba7

imback82 reviewed Mar 21, 2021

View reviewed changes

review commit

04b3b53

sezruby force-pushed the source_extension_timetravel branch from 661f634 to 04b3b53 Compare March 22, 2021 00:04

imback82 reviewed Mar 22, 2021

View reviewed changes

sezruby added 2 commits March 21, 2021 22:08

review commit 333

d0a7fdd

review commit 3333

6598dc3

imback82 approved these changes Mar 22, 2021

View reviewed changes

imback82 merged commit 0843a37 into microsoft:master Mar 22, 2021

andrei-ionescu mentioned this pull request Mar 22, 2021

Add Iceberg version history to IndexLogEntry for efficient time travel query #388

Open

sezruby mentioned this pull request Mar 23, 2021

Check available index versions for Delta Lake time travel query #389

Merged

sezruby deleted the source_extension_timetravel branch April 30, 2021 03:27

sezruby mentioned this pull request Jul 2, 2021

deleteOldVersions API #118

Open

Add Delta Lake version history to IndexLogEntry for efficient time travel query #272

Add Delta Lake version history to IndexLogEntry for efficient time travel query #272

Uh oh!

Conversation

sezruby commented Nov 25, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the context for this pull request?

What changes were proposed in this pull request?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

sezruby commented Dec 10, 2020

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!

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!

imback82 commented Feb 2, 2021

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!

imback82 commented Feb 12, 2021

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

imback82 left a comment

Choose a reason for hiding this comment

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!

sezruby commented Nov 25, 2020 •

edited

Loading