commit

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

@@ -60,6 +60,11 @@ grammar DeltaSqlBase;
       return true;
     }
   }
+
+  /**
+   * When true, double quoted literals are identifiers rather than STRINGs.
+   */
+  public boolean double_quoted_identifiers = false;
 }
 
 tokens {
@@ -91,9 +96,25 @@ statement
         (zorderSpec)?                                                   #optimizeTable
     | SHOW COLUMNS (IN | FROM) tableName=qualifiedName
         ((IN | FROM) schemaName=identifier)?                            #showColumns
+    | cloneTableHeader SHALLOW CLONE source=qualifiedName clause=temporalClause?
+       (TBLPROPERTIES tableProps=propertyList)?
+       (LOCATION location=stringLit)?                                   #clone
     | .*?                                                               #passThrough
     ;
 
+createTableHeader
+    : CREATE TABLE (IF NOT EXISTS)? table=qualifiedName
+    ;
+
+replaceTableHeader
+    : (CREATE OR)? REPLACE TABLE table=qualifiedName
+    ;
+
+cloneTableHeader
+    : createTableHeader
+    | replaceTableHeader
+    ;
+
 zorderSpec
     : ZORDER BY LEFT_PAREN interleave+=qualifiedName (COMMA interleave+=qualifiedName)* RIGHT_PAREN
     | ZORDER BY interleave+=qualifiedName (COMMA interleave+=qualifiedName)*
@@ -108,6 +129,36 @@ qualifiedName
     : identifier ('.' identifier)*
     ;
 
+propertyList
+    : LEFT_PAREN property (COMMA property)* RIGHT_PAREN
+    ;
+
+property
+    : key=propertyKey (EQ? value=propertyValue)?
+    ;
+
+propertyKey
+    : identifier (DOT identifier)*
+    | stringLit
+    ;
+
+propertyValue
+    : INTEGER_VALUE
+    | DECIMAL_VALUE
+    | booleanValue
+    | identifier LEFT_PAREN stringLit COMMA stringLit RIGHT_PAREN
+    | value=stringLit
+    ;
+
+stringLit
+    : STRING
+    | {!double_quoted_identifiers}? DOUBLEQUOTED_STRING
+    ;
+
+booleanValue
+    : TRUE | FALSE
+    ;
+
 identifier
     : IDENTIFIER             #unquotedIdentifier
     | quotedIdentifier       #quotedIdentifierAlternative
@@ -167,6 +218,7 @@ nonReserved
     | RESTORE | AS | OF
     | ZORDER | LEFT_PAREN | RIGHT_PAREN
     | SHOW | COLUMNS | IN | FROM | NO | STATISTICS
+    | CLONE | SHALLOW
     ;
 
 // Define how the keywords above should appear in a user's SQL statement.
@@ -175,18 +227,22 @@ ALTER: 'ALTER';
 AS: 'AS';
 BY: 'BY';
 CHECK: 'CHECK';
+CLONE: 'CLONE';
 COLUMNS: 'COLUMNS';
 COMMA: ',';
 COMMENT: 'COMMENT';
 CONSTRAINT: 'CONSTRAINT';
 CONVERT: 'CONVERT';
+CREATE: 'CREATE';
 DELTA: 'DELTA';
 DESC: 'DESC';
 DESCRIBE: 'DESCRIBE';
 DETAIL: 'DETAIL';
+DOT: '.';
 DROP: 'DROP';
 DRY: 'DRY';
 EXISTS: 'EXISTS';
+FALSE: 'FALSE';
 FOR: 'FOR';
 FROM: 'FROM';
 GENERATE: 'GENERATE';
@@ -196,23 +252,29 @@ IF: 'IF';
 IN: 'IN';
 LEFT_PAREN: '(';
 LIMIT: 'LIMIT';
+LOCATION: 'LOCATION';
 MINUS: '-';
 NO: 'NO';
 NOT: 'NOT' | '!';
 NULL: 'NULL';
 OF: 'OF';
+OR: 'OR';
 OPTIMIZE: 'OPTIMIZE';
 PARTITIONED: 'PARTITIONED';
+REPLACE: 'REPLACE';
 RESTORE: 'RESTORE';
 RETAIN: 'RETAIN';
 RIGHT_PAREN: ')';
 RUN: 'RUN';
+SHALLOW: 'SHALLOW';
 SHOW: 'SHOW';
 SYSTEM_TIME: 'SYSTEM_TIME';
 SYSTEM_VERSION: 'SYSTEM_VERSION';
 TABLE: 'TABLE';
+TBLPROPERTIES: 'TBLPROPERTIES';
 TIMESTAMP: 'TIMESTAMP';
 TO: 'TO';
+TRUE: 'TRUE';
 VACUUM: 'VACUUM';
 VERSION: 'VERSION';
 WHERE: 'WHERE';
@@ -236,6 +298,10 @@ STRING
     | '"' ( ~('"'|'\\') | ('\\' .) )* '"'
     ;
 
+DOUBLEQUOTED_STRING
+    :'"' ( ~('"'|'\\') | ('\\' .) )* '"'
+    ;
+
 BIGINT_LITERAL
     : DIGIT+ 'L'
     ;

core/src/main/resources/error/delta-error-classes.json

@@ -278,6 +278,22 @@
     ],
     "sqlState" : "22000"
   },
+  "DELTA_CLONE_AMBIGUOUS_TARGET" : {
+    "message" : [
+      "",
+      "Two paths were provided as the CLONE target so it is ambiguous which to use. An external",
+      "location for CLONE was provided at <externalLocation> at the same time as the path",
+      "<targetIdentifier>."
+    ],
+    "sqlState" : "42000"
+  },
+  "DELTA_CLONE_UNSUPPORTED_SOURCE" : {
+    "message" : [
+      "Unsupported clone source '<name>', whose format is <format>.",
+      "The supported formats are 'delta' and 'parquet'."
+    ],
+    "sqlState" : "0A000"
+  },
   "DELTA_COLUMN_NOT_FOUND" : {
     "message" : [
       "Unable to find the column `<columnName>` given [<columnList>]"
@@ -731,6 +747,13 @@
     ],
     "sqlState" : "0A000"
   },
+  "DELTA_INVALID_CLONE_PATH" : {
+    "message" : [
+      "The target location for CLONE needs to be an absolute path or table name. Use an",
+      "absolute path instead of <path>."
+    ],
+    "sqlState" : "22000"
+  },
   "DELTA_INVALID_COMMITTED_VERSION" : {
     "message" : [
       "The committed version is <committedVersion> but the current version is <currentVersion>."
@@ -1664,6 +1687,15 @@
     ],
     "sqlState" : "0A000"
   },
+  "DELTA_UNSUPPORTED_CLONE_REPLACE_SAME_TABLE" : {
+    "message" : [
+      "",
+      "You tried to REPLACE an existing table (<tableName>) with CLONE. This operation is",
+      "unsupported. Try a different target for CLONE or delete the table at the current target.",
+      ""
+    ],
+    "sqlState" : "0A000"
+  },
   "DELTA_UNSUPPORTED_COLUMN_MAPPING_MODE_CHANGE" : {
     "message" : [
       "Changing column mapping mode from '<oldMode>' to '<newMode>' is not supported."
@@ -1790,6 +1822,12 @@
     ],
     "sqlState" : "0A000"
   },
+  "DELTA_UNSUPPORTED_NON_EMPTY_CLONE" : {
+    "message" : [
+      "The clone destination table is non-empty. Please TRUNCATE or DELETE FROM the table before running CLONE."
+    ],
+    "sqlState" : "0A000"
+  },
   "DELTA_UNSUPPORTED_OUTPUT_MODE" : {
     "message" : [
       "Data source <dataSource> does not support <mode> output mode"

core/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala

@@ -59,9 +59,11 @@ import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.parser.{ParseErrorListener, ParseException, ParserInterface}
 import org.apache.spark.sql.catalyst.parser.ParserUtils.{string, withOrigin}
-import org.apache.spark.sql.catalyst.plans.logical.{AlterTableAddConstraint, AlterTableDropConstraint, LogicalPlan, RestoreTableStatement}
+import org.apache.spark.sql.catalyst.plans.logical.{AlterTableAddConstraint, AlterTableDropConstraint, CloneTableStatement, LogicalPlan, RestoreTableStatement}
 import org.apache.spark.sql.catalyst.trees.Origin
-import org.apache.spark.sql.internal.VariableSubstitution
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, TableCatalog}
+import org.apache.spark.sql.errors.QueryParsingErrors
+import org.apache.spark.sql.internal.{SQLConf, VariableSubstitution}
 import org.apache.spark.sql.types._
 
 /**
@@ -153,6 +155,147 @@ class DeltaSqlParser(val delegate: ParserInterface) extends ParserInterface {
  */
 class DeltaSqlAstBuilder extends DeltaSqlBaseBaseVisitor[AnyRef] {
 
+  import org.apache.spark.sql.catalyst.parser.ParserUtils._
+
+  /**
+   * Convert a property list into a key-value map.
+   * This should be called through [[visitPropertyKeyValues]] or [[visitPropertyKeys]].
+   */
+  override def visitPropertyList(
+      ctx: PropertyListContext): Map[String, String] = withOrigin(ctx) {
+    val properties = ctx.property.asScala.map { property =>
+      val key = visitPropertyKey(property.key)
+      val value = visitPropertyValue(property.value)
+      key -> value
+    }
+    // Check for duplicate property names.
+    checkDuplicateKeys(properties.toSeq, ctx)
+    properties.toMap
+  }
+
+  /**
+   * Parse a key-value map from a [[PropertyListContext]], assuming all values are specified.
+   */
+  def visitPropertyKeyValues(ctx: PropertyListContext): Map[String, String] = {
+    val props = visitPropertyList(ctx)
+    val badKeys = props.collect { case (key, null) => key }
+    if (badKeys.nonEmpty) {
+      operationNotAllowed(
+        s"Values must be specified for key(s): ${badKeys.mkString("[", ",", "]")}", ctx)
+    }
+    props
+  }
+
+  /**
+   * Parse a list of keys from a [[PropertyListContext]], assuming no values are specified.
+   */
+  def visitPropertyKeys(ctx: PropertyListContext): Seq[String] = {
+    val props = visitPropertyList(ctx)
+    val badKeys = props.filter { case (_, v) => v != null }.keys
+    if (badKeys.nonEmpty) {
+      operationNotAllowed(
+        s"Values should not be specified for key(s): ${badKeys.mkString("[", ",", "]")}", ctx)
+    }
+    props.keys.toSeq
+  }
+
+  /**
+   * A property key can either be String or a collection of dot separated elements. This
+   * function extracts the property key based on whether its a string literal or a property
+   * identifier.
+   */
+  override def visitPropertyKey(key: PropertyKeyContext): String = {
+    if (key.stringLit() != null) {
+      string(visitStringLit(key.stringLit()))
+    } else {
+      key.getText
+    }
+  }
+
+  /**
+   * A property value can be String, Integer, Boolean or Decimal. This function extracts
+   * the property value based on whether its a string, integer, boolean or decimal literal.
+   */
+  override def visitPropertyValue(value: PropertyValueContext): String = {
+    if (value == null) {
+      null
+    } else if (value.identifier != null) {
+      value.identifier.getText
+    } else if (value.value != null) {
+      string(visitStringLit(value.value))
+    } else if (value.booleanValue != null) {
+      value.getText.toLowerCase(Locale.ROOT)
+    } else {
+      value.getText
+    }
+  }
+
+  override def visitStringLit(ctx: StringLitContext): Token = {
+    if (ctx != null) {
+      if (ctx.STRING != null) {
+        ctx.STRING.getSymbol
+      } else {
+        ctx.DOUBLEQUOTED_STRING.getSymbol
+      }
+    } else {
+      null
+    }
+  }
+
+  /**
+   * Parse either create table header or replace table header.
+   * @return TableIdentifier for the target table
+   *         Boolean for whether we are creating a table
+   *         Boolean for whether we are replacing a table
+   *         Boolean for whether we are creating a table if not exists
+   */
+  override def visitCloneTableHeader(
+      ctx: CloneTableHeaderContext): (TableIdentifier, Boolean, Boolean, Boolean) = withOrigin(ctx) {
+    ctx.children.asScala.head match {
+      case createHeader: CreateTableHeaderContext =>
+        (visitTableIdentifier(createHeader.table), true, false, createHeader.EXISTS() != null)
+      case replaceHeader: ReplaceTableHeaderContext =>
+        (visitTableIdentifier(replaceHeader.table), replaceHeader.CREATE() != null, true, false)
+      case _ =>
+        throw new ParseException("Incorrect CLONE header expected REPLACE or CREATE table", ctx)
+    }
+  }
+
+  /**
+   * Creates a [[CloneTableStatement]] logical plan. Example SQL:
+   * {{{
+   *   CREATE [OR REPLACE] TABLE <table-identifier> SHALLOW CLONE <source-table-identifier>
+   *     [TBLPROPERTIES ('propA' = 'valueA', ...)]
+   *     [LOCATION '/path/to/cloned/table']
+   * }}}
+   */
+  override def visitClone(ctx: CloneContext): LogicalPlan = withOrigin(ctx) {
+    val (target, isCreate, isReplace, ifNotExists) = visitCloneTableHeader(ctx.cloneTableHeader())
+
+    if (!isCreate && ifNotExists) {
+      throw new ParseException(
+        "IF NOT EXISTS cannot be used together with REPLACE", ctx.cloneTableHeader())
+    }
+
+    // Get source for clone (and time travel source if necessary)
+    val sourceRelation = UnresolvedRelation(visitTableIdentifier(ctx.source))
+    val maybeTimeTravelSource = maybeTimeTravelChild(ctx.clause, sourceRelation)
+    val targetRelation = UnresolvedRelation(target)
+
+    val tablePropertyOverrides = Option(ctx.tableProps)
+      .map(visitPropertyKeyValues)
+      .getOrElse(Map.empty[String, String])
+
+    CloneTableStatement(
+      maybeTimeTravelSource,
+      targetRelation,
+      ifNotExists,
+      isReplace,
+      isCreate,
+      tablePropertyOverrides,
+      Option(ctx.location).map(s => string(visitStringLit(s))))
+  }
+
   /**
    * Create a [[VacuumTableCommand]] logical plan. Example SQL:
    * {{{