Support SHALLOW CLONE for Delta Lake #1505

jackierwzhang · 2022-12-01T01:25:53Z

Description

This PR introduces support for SHALLOW CLONE for both Delta and Parquet table on Delta Lake, specifically the following command:
CREATE [OR REPLACE] TABLE [IF NOT EXISTS] target SHALLOW CLONE source [VERSION AS OF version | TIMESTAMP AS OF timestamp] [TBLPROPERTIES clause] [LOCATION path]

This enables the following use cases:

Create a target table with Delta log pointing to the files from the source table. The source table can be either a Delta table or a Parquet table.

You may also specify a custom path to create as an external table in a path, a clause for additional table properties to append to, or a version to create the target table as a time-travelled version of the source table (if is Delta).

E.g. CREATE TABLE target SHALLOW CLONE source

Replace/restore a table by shallow-cloning itself. This requires the table to be empty by the time of cloning to avoid data duplication.

E.g. REPLACE TABLE source SHALLOW CLONE source VERSION AS OF 1

How was this patch tested?

Unit tests.

Does this PR introduce any user-facing changes?

No.

core/src/main/antlr4/io/delta/sql/parser/DeltaSqlBase.g4

core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala

core/src/main/antlr4/io/delta/sql/parser/DeltaSqlBase.g4

vkorukanti · 2022-12-03T22:46:19Z

core/src/test/scala/org/apache/spark/sql/delta/CloneTableSQLSuite.scala

+  with DeltaColumnMappingTestUtils
+{
+  // scalastyle:off argcount
+  override protected def cloneTable(


Is there a Scala/Python API planned in addition to the SQL API?

Planned, but not yet available. It can be a follow up.

core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala

vkorukanti · 2022-12-05T06:12:59Z

core/src/main/scala/org/apache/spark/sql/delta/DeltaAnalysis.scala

+          output = CloneTableCommand.output)
+
+      // Non-delta metastore table already exists at target
+      case LogicalRelation(_, _, existingCatalog @ Some(catalog), _) =>


What do we do if there is a non-delta table exists in catalog which is same as the clone target table?

It would depend on the SaveMode, e.g. here

core/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

vkorukanti · 2022-12-05T06:20:43Z

core/src/main/scala/org/apache/spark/sql/delta/commands/CloneTableBase.scala

+  protected def getOutputSeq(operationMetrics: Map[String, Long]): Seq[Row]
+
+
+  protected val fnfExceptionMessage: String = "Tried to clone a version of the table where " +


where is this used? Do we check for the existence of files in source table before the clone?

yeah we do, but this error message is not useful any more.

vkorukanti · 2022-12-05T06:43:14Z

core/src/main/scala/org/apache/spark/sql/delta/commands/CloneTableBase.scala

+    checkColumnMappingMode(newMetadata, metadataToUpdate)
+    // 3. Checks for column mapping mode conflicts with existing metadata if there's any
+    if (txn.readVersion >= 0) {
+      checkColumnMappingMode(txn.snapshot.metadata, metadataToUpdate)


Why do we need to check for column mapping mode when we are replacing the target table? Also is there a test for it, in case we need to check.

Yeah because we allow changing table properties during replace, this checks against that.

vkorukanti

LGTM!

One test for checking the target table history after clone, to make sure all op metrics are correct, seems to be missing. If it exists already, you can ignore the comment. If it is missing, you can post that in a separate PR.

felipepessoto · 2022-12-15T22:06:22Z

@jackierwzhang, @vkorukanti, it seems the config name doesn't follow the same pattern used by recent features, should we change it?

clone.replaceEnabled -> clone.replace.enabled?

Example:

val DELTA_CONVERT_ICEBERG_ENABLED =
buildConf("convert.iceberg.enabled")
.internal()
.doc("If enabled, Iceberg tables can be converted into a Delta table.")
.booleanConf
.createWithDefault(true)

val DELTA_CONVERT_ICEBERG_PARTITION_EVOLUTION_ENABLED =
buildConf("convert.iceberg.partitionEvolution.enabled")
.doc("If enabled, support conversion of iceberg tables experienced partition evolution.")
.booleanConf
.createWithDefault(false)

As a followup to the SHALLOW CLONE [support](#1505) for Delta Lake, it would be great if we could enable SHALLOW CLONE on an Iceberg table as well. This will be a CLONVERT (CLONE + CONVERT) operation, in which we will create a Delta catalog table with files pointing to the original Iceberg table in one transaction. 1. It allows users to quickly experiment with Delta Lake without modifying the original Iceberg table's data. 2. It simplifies the user flow by combining a Delta catalog table creation with an Iceberg conversion. Similar to SHALLOW CLONE, it will work as follows: 1. Clone a Iceberg catalog table (after the setup [here](https://iceberg.apache.org/docs/latest/getting-started/)) ``` CREATE TABLE [IF NOT EXISTS] delta SHALLOW CLONE iceberg.db.table [TBLPROPERTIES clause] [LOCATION path] ``` 2. Clone a path-based Iceberg table ``` CREATE TABLE [IF NOT EXISTS] delta SHALLOW CLONE iceberg.`/path/to/iceberg/table`[TBLPROPERTIES clause] [LOCATION path] ``` Closes #1522 New unit tests. No. GitOrigin-RevId: e01994e037cf44e06f4ef3d6f185f5925dd77e48

felipepessoto · 2023-09-14T05:28:19Z

core/src/main/scala/org/apache/spark/sql/delta/commands/CloneTableCommand.scala

+  }
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    if (!targetPath.isAbsolute) {


@jackierwzhang, do you remember why we need to validate if it is an absolute path?

scottsand-db requested a review from vkorukanti December 1, 2022 17:24

scottsand-db assigned jackierwzhang Dec 1, 2022

scottsand-db self-requested a review December 1, 2022 17:27

commit

9cc139e

jackierwzhang force-pushed the dbr-oss1387 branch from 9c9c76d to 9cc139e Compare December 1, 2022 17:43

retrigger test

0b9f494

vkorukanti reviewed Dec 5, 2022

View reviewed changes

jackierwzhang added 2 commits December 5, 2022 14:43

comments

e011d33

test

b87199d

jackierwzhang requested review from vkorukanti and removed request for scottsand-db December 5, 2022 23:43

comments

de16fc6

vkorukanti approved these changes Dec 6, 2022

View reviewed changes

jackierwzhang mentioned this pull request Dec 8, 2022

[Feature Request] Support SHALLOW CLONVERT on Iceberg Tables #1516

Closed

3 tasks

vkorukanti closed this in b038b9d Dec 12, 2022

jackierwzhang mentioned this pull request Dec 13, 2022

Support SHALLOW CLONE of Iceberg Tables #1522

Closed

felipepessoto reviewed Sep 14, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support SHALLOW CLONE for Delta Lake #1505

Support SHALLOW CLONE for Delta Lake #1505

jackierwzhang commented Dec 1, 2022 •

edited

vkorukanti Dec 3, 2022

jackierwzhang Dec 5, 2022 •

edited

vkorukanti Dec 5, 2022

jackierwzhang Dec 5, 2022

vkorukanti Dec 5, 2022

jackierwzhang Dec 5, 2022

vkorukanti Dec 5, 2022

jackierwzhang Dec 5, 2022

vkorukanti left a comment

felipepessoto commented Dec 15, 2022

felipepessoto Sep 14, 2023

		protected def getOutputSeq(operationMetrics: Map[String, Long]): Seq[Row]


		protected val fnfExceptionMessage: String = "Tried to clone a version of the table where " +

Support SHALLOW CLONE for Delta Lake #1505

Support SHALLOW CLONE for Delta Lake #1505

Conversation

jackierwzhang commented Dec 1, 2022 • edited

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

Choose a reason for hiding this comment

jackierwzhang Dec 5, 2022 • edited

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

vkorukanti left a comment

Choose a reason for hiding this comment

felipepessoto commented Dec 15, 2022

Choose a reason for hiding this comment

jackierwzhang commented Dec 1, 2022 •

edited

jackierwzhang Dec 5, 2022 •

edited