feat(tables): gate RTAS per-table behind 'replace.enabled' table property

apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java

@@ -65,6 +65,9 @@ void testReplaceTransaction() throws Exception {
       Table table = catalog.buildTable(TABLE_IDENT, SCHEMA).withPartitionSpec(SPEC).create();
       table.newAppend().appendFile(FILE_A).commit();
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      table.updateProperties().set("replace.enabled", "true").commit();
+
       String originalLocation = table.location();
       String originalMetadataLocation = getMetadataLocation(table);
       long originalSnapshotId = table.currentSnapshot().snapshotId();
@@ -120,6 +123,9 @@ void testCreateOrReplaceTransaction() throws Exception {
       // verify that the table was created with one snapshot
       assertEquals(1, Iterables.size(table.snapshots()), "Should have one snapshot after create");
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      table.updateProperties().set("replace.enabled", "true").commit();
+
       // create or replace on existing table should replace it
       Transaction replaceTxn =
           catalog

apps/spark/src/test/java/com/linkedin/openhouse/jobs/spark/OperationsTest.java

@@ -752,6 +752,12 @@ public void testSnapshotsExpirationAfterReplaceTable() throws Exception {
       prepareTable(ops, tableName);
       populateTable(ops, tableName, numInserts);
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      ops.spark()
+          .sql(
+              String.format(
+                  "ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName));
+
       // replace the table using RTAS
       ops.spark()
           .sql(

integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/RTASTest.java

@@ -10,6 +10,7 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.BadRequestException;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.Row;
@@ -60,6 +61,10 @@ public void testRTAS() throws Exception {
 
       spark.sql(String.format("ALTER TABLE %s SET POLICY (HISTORY MAX_AGE=24H)", tableName));
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      spark.sql(
+          String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName));
+
       String expectedTableLocation = catalog.loadTable(tableIdent).location();
 
       // replace table
@@ -111,6 +116,29 @@ public void testRTAS() throws Exception {
     }
   }
 
+  @Test
+  public void testRTASFailsWhenReplaceDisabled() throws Exception {
+    try (SparkSession spark = getSparkSession()) {
+      // create the table without opting into RTAS; replace is disabled by default
+      spark.sql(
+          String.format(
+              "CREATE TABLE %s USING iceberg AS SELECT * FROM %s", tableName, sourceName));
+
+      // REPLACE TABLE should be rejected because 'replace.enabled' is not set on the table
+      BadRequestException exception =
+          assertThrows(
+              BadRequestException.class,
+              () ->
+                  spark.sql(
+                      String.format(
+                          "REPLACE TABLE %s USING iceberg AS SELECT * FROM %s",
+                          tableName, sourceName)));
+      assertTrue(
+          exception.getMessage().contains("Replace (RTAS) is not enabled"),
+          "Expected an RTAS-disabled error but got: " + exception.getMessage());
+    }
+  }
+
   @Test
   public void testCreateRTAS() throws Exception {
     try (SparkSession spark = getSparkSession()) {
@@ -125,6 +153,10 @@ public void testCreateRTAS() throws Exception {
 
       String expectedTableLocation = catalog.loadTable(tableIdent).location();
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      spark.sql(
+          String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName));
+
       // create or replace table should replace the table
       spark.sql(
           String.format(
@@ -169,6 +201,10 @@ public void testDataFrameV2Replace() throws Exception {
 
       spark.table(sourceName).writeTo(tableName).using("iceberg").create();
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      spark.sql(
+          String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName));
+
       String expectedTableLocation = catalog.loadTable(tableIdent).location();
 
       spark
@@ -226,6 +262,10 @@ public void testDataFrameV2CreateOrReplace() throws Exception {
           .using("iceberg")
           .createOrReplace();
 
+      // RTAS is disabled by default; opt the table in before replacing it.
+      spark.sql(
+          String.format("ALTER TABLE %s SET TBLPROPERTIES ('replace.enabled'='true')", tableName));
+
       String expectedTableLocation = catalog.loadTable(tableIdent).location();
 
       spark

services/common/src/main/java/com/linkedin/openhouse/common/exception/UnsupportedClientOperationException.java

@@ -20,6 +20,7 @@ public enum Operation {
     GRANT_ON_UNSHARED_TABLES,
     ALTER_TABLE_TYPE,
     GRANT_ON_LOCKED_TABLES,
-    LOCKED_TABLE_OPERATION
+    LOCKED_TABLE_OPERATION,
+    RTAS_ON_DISABLED_TABLE
   }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/services/IcebergSnapshotsServiceImpl.java

@@ -67,6 +67,8 @@ public Pair<TableDto, Boolean> putIcebergSnapshots(
 
     if (tableDto.isPresent()
         && icebergSnapshotRequestBody.getCreateUpdateTableRequestBody().isReplaceCommit()) {
+      // RTAS is gated per-table and disabled by default; ensure it is enabled on the table.
+      authorizationUtils.checkReplaceTableEnabled(tableDto.get());
       // Check if table creator has the privilege to replace the table.
       authorizationUtils.checkReplaceTablePrivilege(tableDto.get(), tableCreatorUpdater);
     } else if (tableDto.isPresent()) {

services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesServiceImpl.java

@@ -100,6 +100,8 @@ public Pair<TableDto, Boolean> putTable(
 
     // Special case handling
     if (tableDto.isPresent() && createUpdateTableRequestBody.isStageReplace()) {
+      // RTAS is gated per-table and disabled by default; ensure it is enabled on the table.
+      authorizationUtils.checkReplaceTableEnabled(tableDto.get());
       // Check if table creator has the privilege to replace the table.
       authorizationUtils.checkReplaceTablePrivilege(tableDto.get(), tableCreatorUpdater);
     } else if (tableDto.isPresent()) {

services/tables/src/main/java/com/linkedin/openhouse/tables/utils/AuthorizationUtils.java

@@ -1,10 +1,12 @@
 package com.linkedin.openhouse.tables.utils;
 
+import com.linkedin.openhouse.common.exception.UnsupportedClientOperationException;
 import com.linkedin.openhouse.tables.authorization.AuthorizationHandler;
 import com.linkedin.openhouse.tables.authorization.Privileges;
 import com.linkedin.openhouse.tables.common.TableType;
 import com.linkedin.openhouse.tables.model.DatabaseDto;
 import com.linkedin.openhouse.tables.model.TableDto;
+import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.security.access.AccessDeniedException;
@@ -15,6 +17,12 @@
 @Component
 public class AuthorizationUtils {
 
+  /**
+   * Table property that gates RTAS (Replace Table As Select). RTAS is disabled by default; a table
+   * can only be replaced when this property is explicitly set to {@code true}.
+   */
+  public static final String RTAS_ENABLED_TABLE_PROP = "replace.enabled";
+
   @Autowired AuthorizationHandler authorizationHandler;
 
   /**
@@ -118,4 +126,25 @@ public void checkReplaceTablePrivilege(TableDto tableDto, String actingPrincipal
               actingPrincipal));
     }
   }
+
+  /**
+   * Checks if RTAS (Replace Table As Select) is enabled for the table. RTAS is disabled by default
+   * and a table can only be replaced once its {@value #RTAS_ENABLED_TABLE_PROP} table property has
+   * been explicitly set to {@code true}. The check is performed against the table's persisted
+   * properties so the gate cannot be bypassed by the incoming request body.
+   *
+   * @param tableDto the existing table targeted by the replace operation
+   */
+  public void checkReplaceTableEnabled(TableDto tableDto) {
+    Map<String, String> tableProperties = tableDto.getTableProperties();
+    if (tableProperties == null
+        || !Boolean.parseBoolean(tableProperties.get(RTAS_ENABLED_TABLE_PROP))) {
+      throw new UnsupportedClientOperationException(
+          UnsupportedClientOperationException.Operation.RTAS_ON_DISABLED_TABLE,
+          String.format(
+              "Replace (RTAS) is not enabled for table %s.%s. Set the table property '%s=true' "
+                  + "before issuing a replace.",
+              tableDto.getDatabaseId(), tableDto.getTableId(), RTAS_ENABLED_TABLE_PROP));
+    }
+  }
 }

services/tables/src/test/java/com/linkedin/openhouse/tables/e2e/h2/SnapshotsControllerTest.java

@@ -20,6 +20,7 @@
 import com.linkedin.openhouse.tables.model.TableDto;
 import com.linkedin.openhouse.tables.model.TableDtoPrimaryKey;
 import com.linkedin.openhouse.tables.repository.OpenHouseInternalRepository;
+import com.linkedin.openhouse.tables.utils.AuthorizationUtils;
 import com.linkedin.openhouse.tablestest.annotation.CustomParameterResolver;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -312,6 +313,12 @@ public void testPutSnapshotsReplicaTableType(GetTableResponseBody getTableRespon
   @MethodSource("responseBodyFeeder")
   public void testPutSnapshotsReplaceCommit(GetTableResponseBody getTableResponseBody)
       throws Exception {
+    // RTAS is disabled by default; enable it via the table property so the replace commit is
+    // allowed.
+    Map<String, String> propsWithRtas = new HashMap<>(getTableResponseBody.getTableProperties());
+    propsWithRtas.put(AuthorizationUtils.RTAS_ENABLED_TABLE_PROP, "true");
+    getTableResponseBody = getTableResponseBody.toBuilder().tableProperties(propsWithRtas).build();
+
     // Step 1: Create a table
     MvcResult createResult =
         RequestAndValidateHelper.createTableAndValidateResponse(

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

@@ -47,6 +47,7 @@
 import com.linkedin.openhouse.tables.repository.OpenHouseInternalRepository;
 import com.linkedin.openhouse.tables.toggle.model.TableToggleStatus;
 import com.linkedin.openhouse.tables.toggle.repository.ToggleStatusesRepository;
+import com.linkedin.openhouse.tables.utils.AuthorizationUtils;
 import io.micrometer.core.instrument.search.MeterNotFoundException;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import java.util.ArrayList;
@@ -829,8 +830,14 @@ public void testCreateTableWithIncorrectTableTypeThrowsException() throws Except
   @SneakyThrows
   @Test
   public void testStagedReplace() {
-    GetTableResponseBody stageReplaceTable =
+    GetTableResponseBody baseTable =
         TableModelConstants.buildGetTableResponseBodyWithDbTbl("d_sr", "t_sr");
+    // RTAS is disabled by default; enable it via the table property so the staged replace is
+    // allowed.
+    Map<String, String> propsWithRtas = new HashMap<>(baseTable.getTableProperties());
+    propsWithRtas.put(AuthorizationUtils.RTAS_ENABLED_TABLE_PROP, "true");
+    GetTableResponseBody stageReplaceTable =
+        baseTable.toBuilder().tableProperties(propsWithRtas).build();
 
     // Create a real table first
     MvcResult createResult =
@@ -904,6 +911,38 @@ public void testStagedReplace() {
     RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, stageReplaceTable);
   }
 
+  @SneakyThrows
+  @Test
+  public void testStagedReplaceFailsWhenRtasDisabled() {
+    // Table created without enabling RTAS; replace is disabled by default.
+    GetTableResponseBody table =
+        TableModelConstants.buildGetTableResponseBodyWithDbTbl("d_sr_dis", "t_sr_dis");
+    MvcResult createResult =
+        RequestAndValidateHelper.createTableAndValidateResponse(table, mvc, storageManager);
+    String originalTableLocation =
+        JsonPath.read(createResult.getResponse().getContentAsString(), "$.tableLocation");
+
+    // A stageReplace against a table without replaceEnabled should be rejected.
+    mvc.perform(
+            MockMvcRequestBuilders.post(
+                    String.format(
+                        ValidationUtilities.CURRENT_MAJOR_VERSION_PREFIX + "/databases/%s/tables/",
+                        table.getDatabaseId()))
+                .contentType(MediaType.APPLICATION_JSON)
+                .content(
+                    buildCreateUpdateTableRequestBody(table)
+                        .toBuilder()
+                        .baseTableVersion(originalTableLocation)
+                        .stageReplace(true)
+                        .build()
+                        .toJson())
+                .accept(MediaType.APPLICATION_JSON))
+        .andExpect(status().isBadRequest())
+        .andExpect(jsonPath("$.message", containsString("Replace (RTAS) is not enabled")));
+
+    RequestAndValidateHelper.deleteTableAndValidateResponse(mvc, table);
+  }
+
   @SneakyThrows
   @Test
   public void testStagedCreateDoesntExistInConsecutiveCalls() {