[Spark] Support external DSV2 catalog in RESTORE & Vacuum command #2036

gengliangwang · 2023-09-09T04:25:51Z

Which Delta project/connector is this regarding?

Description

Support external DSV2 catalog in RESTORE & Vacuum command. After the changes, the restore command supports tables from external DSV2 catalog.
For example, with

spark.sql.catalog.customer_catalog=org.apache.spark.sql.CustomerCatalog

We can query

SET CATALOG customer_catalog;
RESTORE TABLE t1 VERSION AS OF 0
VACUUM t1

Or simply

RESTORE TABLE customer_catalog.default.t1 VERSION AS OF 0
VACUUM customer_catalog.default.t1

-->

How was this patch tested?

new end-to-end tests
new parser test cases

Does this PR introduce any user-facing changes?

Yes, users can use RESTORE & Vacuum command on the tables of their external DSV2 catalogs.

gengliangwang · 2023-09-09T04:26:03Z

cc @ryan-johnson-databricks

gengliangwang · 2023-09-09T04:28:28Z

Note: this PR is based on #2033.
To make it easier for review, we can merge #2033 first.
With refactoring, this PR contains a narrow waist for resolving the delta identifier: resolveDeltaIdentifier

gengliangwang · 2023-09-10T21:58:31Z

spark/src/main/scala/org/apache/spark/sql/delta/PreprocessTimeTravel.scala

-          tt.timestamp,
-          tt.version,
-          tt.creationSource))
-
    case ct @ CloneTableStatement(


We can move the resolution of CloneTableStatement to ResolveDeltaIdentifier as well.

gengliangwang · 2023-09-11T17:12:15Z

spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala

@@ -498,8 +498,8 @@ class DeltaVacuumSuite
        val e = intercept[AnalysisException] {
          vacuumSQLTest(tablePath, viewName)
        }
-        assert(e.getMessage.contains("not found") ||
-          e.getMessage.contains("TABLE_OR_VIEW_NOT_FOUND"))
+        assert(e.getMessage.contains("VACUUM is only supported for Delta tables.") ||


The error message is better now when vacuuming on a view.

spark/src/main/scala/org/apache/spark/sql/delta/ResolveDeltaIdentifier.scala

ryan-johnson-databricks · 2023-09-11T14:26:37Z

spark/src/main/scala/org/apache/spark/sql/delta/ResolveDeltaIdentifier.scala

+    val optionalDeltaTableV2 = resolveAsTable(unresolvedId)
+
+    // If the identifier is not a Delta table, try to resolve it as a Delta file table.
+    val deltaTableV2 = optionalDeltaTableV2.getOrElse {


Under what circumstances would resolveAsTable return None for a valid (path-based?) Delta table?
Or, put another way, what circumstances would mean ResolveRelations fails to handle an UnresolvedTable that references a path-based Delta table?

If such circumstances exist, can/should we fix table resolution directly, instead of compensating here?

Yes, it is mentioned in the comment of resolveAsTable. The short answer is that the table identifier doesn't exist in spark catalogs.

// Resolve the identifier as a table in the Spark catalogs. // Return None if the table doesn't exist. // If the table exists but is not a Delta table, throw an exception. // If the identifier is a view, throw an exception.

If such circumstances exist, can/should we fix table resolution directly, instead of compensating here?

What do you mean by "fix table resolution directly"?
I believe there is a corner case here.
In Spark, You can create a table /tmp/foo under a database delta

use delta; create table `/tmp/foo` using delta;

So we always have to look up as a catalog table, and try loading it as a path if the table doesn't exist.

ryan-johnson-databricks · 2023-09-11T14:28:38Z

spark/src/main/scala/org/apache/spark/sql/delta/ResolveDeltaIdentifier.scala

+  // If the identifier is a view, throw an exception.
+  private def resolveAsTable(unresolvedId: UnresolvedDeltaIdentifier): Option[DeltaTableV2] = {
+    val unresolvedTable =
+      UnresolvedTable(unresolvedId.nameParts, unresolvedId.commandName, None)


Wasn't this an UnresolvedRelation before? Are we changing it to UnresolvedTable intentionally?
If so, why needed, and what are the side effects of that change?

In this PR, I introduce a new logical plan UnresolvedDeltaIdentifier. I didn't make it UnresolvedRelation because of resolution order matters for the corner case I mentioned above #2036 (comment) .

The code here coverts a UnresolvedDeltaIdentifier to UnresolvedTable and uses the Spark analyzer rule to resolve the relation.

ryan-johnson-databricks · 2023-09-11T14:43:52Z

spark/src/main/scala/io/delta/tables/execution/VacuumTableCommand.scala

@@ -32,28 +32,14 @@ import org.apache.spark.sql.types.StringType
 * }}}
 */
 case class VacuumTableCommand(
-    path: Option[String],
-    table: Option[TableIdentifier],
+    pathToVacuum: Path,


Can't this just be a RunnableCommand with UnaryLike, whose child starts out as an UnresolvedDeltaPathOrIdentifier the sql parser creates?
(see e.g. how OPTIMIZE command was recently upgraded to handle this situation -- the sql parser just has to call that helper method)

If we go that route, we won't need the VacuumTableStatement any more.

If it doesn't work for some reason, then I suspect OptimizeCommand will also need whatever fix we come up with in this PR?

As I took a look at the implementation of VacuumTableCommand, it only cares about the pathToVacuum.
I didn't use UnresolvedDeltaPathOrIdentifier since I introduced UnresolvedDeltaIdentifier in this PR.

…entifier.scala Co-authored-by: Ryan Johnson <[email protected]>

gengliangwang · 2023-09-11T17:47:26Z

@ryan-johnson-databricks Thanks for the review. We need to decide whether we need to take care of the corner case I mentioned in the PR comment.
If we do, we need a new unresolved logical plan, otherwise, we can’t ensure the resolution order of table(spark) and path table(delta).
If we don’t, let’s simply try resolving any unresolved relation as path table in Delta. But it can be regression in case someone is using a file path as a table name.…

gengliangwang added 12 commits September 7, 2023 16:01

workable version

d1c8379

fix temp view

013f964

add CustomCatalogSuite.scala

90b616b

move resolveCatalogAndIdentifier to DeltaTableUtils

be93ff5

fix style

32e66aa

fix style in DeltaTable.scala

01e7d79

remove unnecessary changes; add one comment

c0baf09

save for now

4a24cdf

workable version

7e2c602

add end-to-end test case

75d9198

add new parser test case

6cb5c24

remove unnecessary code

aa886d4

gengliangwang added 2 commits September 10, 2023 09:57

fix test failures

76791c5

move resolution to a new rule

1f62cac

gengliangwang commented Sep 10, 2023

View reviewed changes

gengliangwang added 3 commits September 10, 2023 20:57

avoid calling loadTable twice

16688f8

use Spark analyzer rule to resolve relation

2134121

error message for restoring a view

e513a4f

gengliangwang mentioned this pull request Sep 11, 2023

[Spark] Support external DSV2 catalog in RESTORE command #2033

Closed

5 tasks

fix test failure

f81cfd6

gengliangwang commented Sep 11, 2023

View reviewed changes

ryan-johnson-databricks reviewed Sep 11, 2023

View reviewed changes

Update spark/src/main/scala/org/apache/spark/sql/delta/ResolveDeltaId…

240cb1b

…entifier.scala Co-authored-by: Ryan Johnson <[email protected]>

gengliangwang closed this Sep 15, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Spark] Support external DSV2 catalog in RESTORE & Vacuum command #2036

[Spark] Support external DSV2 catalog in RESTORE & Vacuum command #2036

gengliangwang commented Sep 9, 2023 •

edited

Loading

gengliangwang commented Sep 9, 2023

gengliangwang commented Sep 9, 2023 •

edited

Loading

gengliangwang Sep 10, 2023

gengliangwang Sep 11, 2023

ryan-johnson-databricks Sep 11, 2023

gengliangwang Sep 11, 2023

gengliangwang Sep 11, 2023

gengliangwang Sep 11, 2023

ryan-johnson-databricks Sep 11, 2023

gengliangwang Sep 11, 2023

ryan-johnson-databricks Sep 11, 2023

gengliangwang Sep 11, 2023

gengliangwang commented Sep 11, 2023

[Spark] Support external DSV2 catalog in RESTORE & Vacuum command #2036

[Spark] Support external DSV2 catalog in RESTORE & Vacuum command #2036

Conversation

gengliangwang commented Sep 9, 2023 • edited Loading

Which Delta project/connector is this regarding?

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

gengliangwang commented Sep 9, 2023

gengliangwang commented Sep 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gengliangwang commented Sep 11, 2023

gengliangwang commented Sep 9, 2023 •

edited

Loading

gengliangwang commented Sep 9, 2023 •

edited

Loading