SQL support for unsetting replication policy (#297)

## Summary

Replication policy allows users to set configs for table replication.
While updates can be done on it, there is not provision to delete the
config in case users want to disable the replication process.

This PR introduces SQL support for purging the replication policy.
Syntax:
`ALTER table <db>.<table> UNSET policy (REPLICATION)`

The syntax is inline with iceberg support for unsetting the table
properties.
ref: https://iceberg.apache.org/docs/1.5.1/spark-ddl/#alter-table

## Changes

- [x] Client-facing API Changes
- [ ] Internal API Changes
- [ ] Bug Fixes
- [x] New Features
- [ ] Performance Improvements
- [ ] Code Style
- [ ] Refactoring
- [ ] Documentation
- [ ] Tests

For all the boxes checked, please include additional details of the
changes made in this pull request.

## Testing Done

Docker testing steps:
1. Create table with replication, retention policy using alter table
statement
2. Unset replication policy. Show tblProperties has policy Json as:
```
|policies                                  |{
  "retention": {
    "count": 20,
    "granularity": "DAY",
    "columnPattern": {
      "columnName": "name",
      "pattern": "'yyyy-MM-dd-HH-mm'"
    }
  },
  "sharingEnabled": false,
  "replication": {
    "config": []
  }
}|
```
3. Set table replication again. Check tblProperties: 
```
|policies                                  |{
  "retention": {
    "count": 20,
    "granularity": "DAY",
    "columnPattern": {
      "columnName": "name",
      "pattern": "'yyyy-MM-dd'"
    }
  },
  "sharingEnabled": false,
  "replication": {
    "config": [
      {
        "destination": "'WAR'",
        "interval": "12H",
        "cronSchedule": "0 30 13/12 ? * * *"
      }
    ]
  }
}|
``` 
4. Update retention policy to validate any other policy can be updated
as expected.

<!--- Check any relevant boxes with "x" -->

- [ ] Manually Tested on local docker setup. Please include commands
ran, and their output.
- [x] Added new tests for the changes made.
- [x] Updated existing tests to reflect the changes made.
- [ ] No tests added or updated. Please explain why. If unsure, please
feel free to ask for help.
- [ ] Some other form of testing like staging or soak time in
production. Please explain.

For all the boxes checked, include a detailed description of the testing
done for the changes made in this pull request.

# Additional Information

- [ ] Breaking Changes
- [ ] Deprecations
- [ ] Large PR broken into smaller PRs, and PR plan linked in the
description.

For all the boxes checked, include additional details of the changes
made in this pull request.

Author: rohitkum2506

Diff for: integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/CatalogOperationTest.java

@@ -1,10 +1,15 @@
 package com.linkedin.openhouse.spark.catalogtest;
 
 import com.google.common.collect.Sets;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.linkedin.openhouse.gen.tables.client.model.Policies;
 import com.linkedin.openhouse.tablestest.OpenHouseSparkITest;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
@@ -15,6 +20,7 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.types.Types;
+import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.types.DateType;
 import org.apache.spark.sql.types.StructField;
@@ -127,6 +133,58 @@ public void testCreateReplicaSkipFieldIdReassignmentUnPartitionedTable() throws
     }
   }
 
+  @Test
+  public void testAlterTableUnsetReplicationPolicy() throws Exception {
+    try (SparkSession spark = getSparkSession()) {
+      spark.sql("CREATE TABLE openhouse.d1.`ttt1` (name string)");
+      spark.sql("INSERT INTO openhouse.d1.ttt1 VALUES ('foo')");
+      spark.sql(
+          "ALTER TABLE openhouse.d1.ttt1 SET POLICY (REPLICATION=({destination:'WAR', interval:12h}))");
+      spark.sql(
+          "ALTER TABLE openhouse.d1.ttt1 SET POLICY (RETENTION= 30d on column name where pattern='yyyy-MM-dd')");
+      Policies policies = getPoliciesObj("openhouse.d1.ttt1", spark);
+      Assertions.assertNotNull(policies);
+      Assertions.assertEquals(
+          "'WAR'", policies.getReplication().getConfig().get(0).getDestination());
+      Assertions.assertNotNull(policies.getRetention());
+      Assertions.assertEquals(
+          "'yyyy-MM-dd'", policies.getRetention().getColumnPattern().getPattern());
+
+      // unset replication policy
+      spark.sql("ALTER TABLE openhouse.d1.ttt1 UNSET POLICY (REPLICATION)");
+      Policies updatedPolicy = getPoliciesObj("openhouse.d1.ttt1", spark);
+      Assertions.assertEquals(updatedPolicy.getReplication().getConfig().size(), 0);
+      // assert that other policies, retention is not modified after unsetting replication
+      Assertions.assertNotNull(updatedPolicy.getRetention());
+      Assertions.assertEquals(
+          "'yyyy-MM-dd'", updatedPolicy.getRetention().getColumnPattern().getPattern());
+
+      // assert retention can be set after unsetting replication
+      spark.sql(
+          "ALTER TABLE openhouse.d1.ttt1 SET POLICY (RETENTION = 30D on COLUMN name WHERE pattern = 'yyyy')");
+      Policies policyWithRetention = getPoliciesObj("openhouse.d1.ttt1", spark);
+      Assertions.assertNotNull(policyWithRetention);
+      Assertions.assertEquals(
+          "'yyyy'", policyWithRetention.getRetention().getColumnPattern().getPattern());
+      Assertions.assertEquals(0, policyWithRetention.getReplication().getConfig().size());
+
+      // assert replication can be set again after retention policy
+      spark.sql(
+          "ALTER TABLE openhouse.d1.ttt1 SET POLICY (REPLICATION=({destination:'WAR', interval:12h}))");
+      Policies policyWithReplication = getPoliciesObj("openhouse.d1.ttt1", spark);
+      Assertions.assertNotNull(policyWithReplication);
+      Assertions.assertEquals(
+          "'WAR'", policyWithReplication.getReplication().getConfig().get(0).getDestination());
+
+      // UNSET policy for table without replication
+      spark.sql("CREATE TABLE openhouse.d1.`tttest1` (name string)");
+      spark.sql("INSERT INTO openhouse.d1.tttest1 VALUES ('foo')");
+      spark.sql("ALTER TABLE openhouse.d1.tttest1 UNSET POLICY (REPLICATION)");
+      Policies policytttest1 = getPoliciesObj("openhouse.d1.tttest1", spark);
+      Assertions.assertEquals(0, policytttest1.getReplication().getConfig().size());
+    }
+  }
+
   @Test
   public void testCreateReplicaSkipFieldIdReassignmentPartitionedTable() throws Exception {
     try (SparkSession spark = getSparkSession()) {
@@ -185,4 +243,13 @@ private Catalog getOpenHouseCatalog(SparkSession spark) {
         catalogProperties,
         spark.sparkContext().hadoopConfiguration());
   }
+
+  private Policies getPoliciesObj(String tableName, SparkSession spark) {
+    List<Row> props = spark.sql(String.format("show tblProperties %s", tableName)).collectAsList();
+    Map<String, String> collect =
+        props.stream().collect(Collectors.toMap(r -> r.getString(0), r -> r.getString(1)));
+    String policiesStr = String.valueOf(collect.get("policies"));
+    Gson gson = new GsonBuilder().setPrettyPrinting().create();
+    return gson.fromJson(policiesStr, Policies.class);
+  }
 }

Diff for: integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/SetTableReplicationPolicyStatementTest.java

@@ -60,6 +60,13 @@ public void testSimpleSetReplicationPolicy() {
     assert isPlanValid(ds, replicationConfigJson);
   }
 
+  @Test
+  public void testSimpleUnSetReplicationPolicy() {
+    String replicationConfigJson = "{replication: null}";
+    Dataset<Row> ds = spark.sql("ALTER TABLE openhouse.db.table UNSET POLICY (REPLICATION)");
+    assert isUnSetPlanValid(ds, replicationConfigJson);
+  }
+
   @Test
   public void testSimpleSetReplicationPolicyOptionalInterval() {
     // Test with optional interval
@@ -241,4 +248,11 @@ private boolean isPlanValid(Dataset<Row> dataframe, String replicationConfigJson
     }
     return isValid;
   }
+
+  @SneakyThrows
+  private boolean isUnSetPlanValid(Dataset<Row> dataframe, String replicationConfigJson) {
+    String queryStr = dataframe.queryExecution().explainString(ExplainMode.fromString("simple"));
+    JsonObject json = new Gson().fromJson(replicationConfigJson, JsonObject.class);
+    return queryStr.contains("REPLICATION") && json.has("replication");
+  }
 }

Diff for: integrations/spark/spark-3.1/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4

@@ -25,6 +25,7 @@ singleStatement
 statement
   : ALTER TABLE multipartIdentifier SET POLICY '(' retentionPolicy (columnRetentionPolicy)? ')'        #setRetentionPolicy
   | ALTER TABLE multipartIdentifier SET POLICY '(' replicationPolicy ')'                               #setReplicationPolicy
+  | ALTER TABLE multipartIdentifier UNSET POLICY '(' replication ')'                                   #unSetReplicationPolicy
   | ALTER TABLE multipartIdentifier SET POLICY '(' sharingPolicy ')'                                   #setSharingPolicy
   | ALTER TABLE multipartIdentifier SET POLICY '(' historyPolicy ')'                                   #setHistoryPolicy
   | ALTER TABLE multipartIdentifier MODIFY columnNameClause SET columnPolicy                           #setColumnPolicyTag
@@ -86,8 +87,12 @@ columnRetentionPolicy
     : ON columnNameClause (columnRetentionPolicyPatternClause)?
     ;
 
+replication
+    : REPLICATION
+    ;
+
 replicationPolicy
-    : REPLICATION '=' tableReplicationPolicy
+    : replication '=' tableReplicationPolicy
     ;
 
 tableReplicationPolicy
@@ -170,6 +175,7 @@ versions
 ALTER: 'ALTER';
 TABLE: 'TABLE';
 SET: 'SET';
+UNSET: 'UNSET';
 POLICY: 'POLICY';
 RETENTION: 'RETENTION';
 REPLICATION: 'REPLICATION';

Diff for: integrations/spark/spark-3.1/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala

@@ -67,6 +67,7 @@ class OpenhouseSparkSqlExtensionsParser (delegate: ParserInterface) extends Pars
       .trim()
     (normalized.startsWith("alter table") &&
       (normalized.contains("set policy")) ||
+      (normalized.contains("unset policy")) ||
       (normalized.contains("modify column") &&
         normalized.contains("set tag"))) ||
       normalized.startsWith("grant") ||

Diff for: integrations/spark/spark-3.1/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala

@@ -2,7 +2,7 @@ package com.linkedin.openhouse.spark.sql.catalyst.parser.extensions
 
 import com.linkedin.openhouse.spark.sql.catalyst.enums.GrantableResourceTypes
 import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseSqlExtensionsParser._
-import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement}
+import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy}
 import com.linkedin.openhouse.spark.sql.catalyst.enums.GrantableResourceTypes.GrantableResourceType
 import com.linkedin.openhouse.gen.tables.client.model.TimePartitionSpec
 import org.antlr.v4.runtime.tree.ParseTree
@@ -33,6 +33,12 @@ class OpenhouseSqlExtensionsAstBuilder (delegate: ParserInterface) extends Openh
     SetReplicationPolicy(tableName, replicationPolicies)
   }
 
+  override def visitUnSetReplicationPolicy(ctx: UnSetReplicationPolicyContext): UnSetReplicationPolicy = {
+    val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier)
+    val replicationPolicies = typedVisit[String](ctx.replication())
+    UnSetReplicationPolicy(tableName, replicationPolicies)
+  }
+
   override def visitSetSharingPolicy(ctx: SetSharingPolicyContext): SetSharingPolicy = {
     val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier)
     val sharing = typedVisit[String](ctx.sharingPolicy())
@@ -132,6 +138,11 @@ class OpenhouseSqlExtensionsAstBuilder (delegate: ParserInterface) extends Openh
     ctx.retentionColumnPatternClause().STRING().getText
   }
 
+  override def visitReplication(ctx: ReplicationContext): String =
+    {
+      ctx.REPLICATION().getText
+    }
+
   override def visitSharingPolicy(ctx: SharingPolicyContext): String = {
     ctx.BOOLEAN().getText
   }

Diff for: integrations/spark/spark-3.1/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/UnSetReplicationPolicy.scala

@@ -0,0 +1,9 @@
+package com.linkedin.openhouse.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.plans.logical.Command
+
+case class UnSetReplicationPolicy(tableName: Seq[String], replicationPolicies: String) extends Command {
+  override def simpleString(maxFields: Int): String = {
+    s"UnSetReplicationPolicy: ${tableName}"
+  }
+}

Diff for: integrations/spark/spark-3.1/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala

@@ -1,6 +1,6 @@
 package com.linkedin.openhouse.spark.sql.execution.datasources.v2
 
-import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement}
+import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy}
 import org.apache.iceberg.spark.{Spark3Util, SparkCatalog, SparkSessionCatalog}
 import org.apache.spark.sql.{SparkSession, Strategy}
 import org.apache.spark.sql.catalyst.expressions.PredicateHelper
@@ -17,6 +17,8 @@ case class OpenhouseDataSourceV2Strategy(spark: SparkSession) extends Strategy w
       SetRetentionPolicyExec(catalog, ident, granularity, count, colName, colPattern) :: Nil
     case SetReplicationPolicy(CatalogAndIdentifierExtractor(catalog, ident), replicationPolicies) =>
       SetReplicationPolicyExec(catalog, ident, replicationPolicies) :: Nil
+    case UnSetReplicationPolicy(CatalogAndIdentifierExtractor(catalog, ident), replicationPolicies) =>
+      UnSetReplicationPolicyExec(catalog, ident, replicationPolicies) :: Nil
     case SetHistoryPolicy(CatalogAndIdentifierExtractor(catalog, ident), granularity, maxAge, versions) =>
       SetHistoryPolicyExec(catalog, ident, granularity, maxAge, versions) :: Nil
     case SetSharingPolicy(CatalogAndIdentifierExtractor(catalog, ident), sharing) =>

Diff for: integrations/spark/spark-3.1/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/UnSetReplicationPolicyExec.scala

@@ -0,0 +1,26 @@
+package com.linkedin.openhouse.spark.sql.execution.datasources.v2
+
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.v2.V2CommandExec
+
+case class UnSetReplicationPolicyExec(catalog: TableCatalog, ident: Identifier, replication: (String)) extends V2CommandExec{
+  override protected def run(): Seq[InternalRow] = {
+    catalog.loadTable(ident) match {
+      case iceberg: SparkTable if iceberg.table().properties().containsKey("openhouse.tableId") =>
+        val key = "updated.openhouse.policy"
+        val value = s"""{"replication": {}}"""
+        iceberg.table().updateProperties()
+          .set(key, value)
+          .commit()
+
+      case table =>
+        throw new UnsupportedOperationException(s"Cannot unset replication policy for non-Openhouse table: $table")
+    }
+    Nil
+  }
+
+  override def output: Seq[Attribute] = Nil
+}

Diff for: integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala

@@ -67,6 +67,7 @@ class OpenhouseSparkSqlExtensionsParser (delegate: ParserInterface) extends Pars
       .trim()
     (normalized.startsWith("alter table") &&
       (normalized.contains("set policy")) ||
+      (normalized.contains("unset policy")) ||
       (normalized.contains("modify column") &&
         normalized.contains("set tag"))) ||
       normalized.startsWith("grant") ||

Diff for: integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/UnSetReplicationPolicy.scala

@@ -0,0 +1,9 @@
+package com.linkedin.openhouse.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.plans.logical.LeafCommand
+
+case class UnSetReplicationPolicy(tableName: Seq[String], replicationPolicies: String) extends LeafCommand {
+  override def simpleString(maxFields: Int): String = {
+    s"UnSetReplicationPolicy: ${tableName}"
+  }
+}

Diff for: integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/UnSetReplicationPolicyExec.scala

@@ -0,0 +1,26 @@
+package com.linkedin.openhouse.spark.sql.execution.datasources.v2
+
+import org.apache.iceberg.spark.source.SparkTable
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec
+
+case class UnSetReplicationPolicyExec(catalog: TableCatalog, ident: Identifier, replication: (String)) extends LeafV2CommandExec{
+  override protected def run(): Seq[InternalRow] = {
+    catalog.loadTable(ident) match {
+      case iceberg: SparkTable if iceberg.table().properties().containsKey("openhouse.tableId") =>
+        val key = "updated.openhouse.policy"
+        val value = s"""{"replication": {}}"""
+        iceberg.table().updateProperties()
+          .set(key, value)
+          .commit()
+
+      case table =>
+        throw new UnsupportedOperationException(s"Cannot unset replication policy for non-Openhouse table: $table")
+    }
+    Nil
+  }
+
+  override def output: Seq[Attribute] = Nil
+}

Diff for: services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/TablesControllerTest.java

@@ -1063,6 +1063,30 @@ public void testUpdateSucceedsForReplicationConfig() throws Exception {
     Assertions.assertTrue(
         RequestAndValidateHelper.validateCronSchedule(updatedReplication.get("cronSchedule")));
 
+    Replication nullReplication = Replication.builder().config(new ArrayList<>()).build();
+    Policies newPoliciesNullRepl = Policies.builder().replication(nullReplication).build();
+
+    GetTableResponseBody newContainer =
+        GetTableResponseBody.builder().policies(newPoliciesNullRepl).build();
+    GetTableResponseBody addNullProp = buildGetTableResponseBody(mvcResult, newContainer);
+    mvcResult =
+        mvc.perform(
+                MockMvcRequestBuilders.put(
+                        String.format(
+                            ValidationUtilities.CURRENT_MAJOR_VERSION_PREFIX
+                                + "/databases/%s/tables/%s",
+                            addProp.getDatabaseId(),
+                            addProp.getTableId()))
+                    .contentType(MediaType.APPLICATION_JSON)
+                    .content(buildCreateUpdateTableRequestBody(addNullProp).toJson())
+                    .accept(MediaType.APPLICATION_JSON))
+            .andExpect(status().isOk())
+            .andReturn();
+
+    LinkedHashMap<String, String> updatedNullReplication =
+        JsonPath.read(mvcResult.getResponse().getContentAsString(), "$.policies.replication");
+
+    Assertions.assertTrue(updatedNullReplication.containsKey("config"));
     RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, GET_TABLE_RESPONSE_BODY);
   }