diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
index a683533473be..4ede53bd165f 100644
--- a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
@@ -192,6 +192,14 @@ public Table create() {
 
       String baseLocation = location != null ? location : defaultWarehouseLocation(identifier);
       tableProperties.putAll(tableOverrideProperties());
+
+      if (Boolean.parseBoolean(tableProperties.get(TableProperties.UNIQUE_LOCATION))) {
+        boolean alreadyExists = ops.io().newInputFile(baseLocation).exists();
+        if (alreadyExists) {
+          throw new AlreadyExistsException("Table location already in use: %s", baseLocation);
+        }
+      }
+
       TableMetadata metadata =
           TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, tableProperties);
 
diff --git a/core/src/main/java/org/apache/iceberg/TableProperties.java b/core/src/main/java/org/apache/iceberg/TableProperties.java
index a9116bc57f83..cb53934fb801 100644
--- a/core/src/main/java/org/apache/iceberg/TableProperties.java
+++ b/core/src/main/java/org/apache/iceberg/TableProperties.java
@@ -365,4 +365,7 @@ private TableProperties() {}
 
   public static final String UPSERT_ENABLED = "write.upsert.enabled";
   public static final boolean UPSERT_ENABLED_DEFAULT = false;
+
+  public static final String UNIQUE_LOCATION = "location.unique";
+  public static final boolean UNIQUE_LOCATION_DEFAULT = false;
 }
diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
index 4f889b24cae8..c9d21c46217b 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
@@ -20,6 +20,7 @@
 
 import static org.apache.iceberg.NullOrder.NULLS_FIRST;
 import static org.apache.iceberg.SortDirection.ASC;
+import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
@@ -36,6 +37,7 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -202,6 +204,31 @@ public void testBasicCatalog() throws Exception {
     Assertions.assertThat(fs.isDirectory(new Path(metaLocation))).isFalse();
   }
 
+  @Test
+  public void testCreateTableWithUniqueLocation() throws IOException {
+    TableIdentifier tableIdent = TableIdentifier.of("db", "ns1", "ns2", "unique");
+    ImmutableMap<String, String> catalogProps =
+        ImmutableMap.of(String.format("table-default.%s", TableProperties.UNIQUE_LOCATION), "true");
+
+    HadoopCatalog catalog = hadoopCatalog(catalogProps);
+    String location = catalog.defaultWarehouseLocation(tableIdent);
+    FileSystem fs = Util.getFs(new Path(location), catalog.getConf());
+    Assertions.assertThat(fs.mkdirs(new Path(location))).isTrue();
+
+    Assertions.assertThatThrownBy(
+            () -> catalog.buildTable(tableIdent, SCHEMA).withPartitionSpec(SPEC).create())
+        .isInstanceOf(AlreadyExistsException.class)
+        .hasMessageStartingWith("Table location already in use");
+
+    Table recreated =
+        catalog
+            .buildTable(tableIdent, SCHEMA)
+            .withProperty(TableProperties.UNIQUE_LOCATION, "false")
+            .create();
+    assertThat(recreated.location()).isEqualTo(location);
+    catalog.dropTable(tableIdent);
+  }
+
   @Test
   public void testCreateAndDropTableWithoutNamespace() throws Exception {
     HadoopCatalog catalog = hadoopCatalog();
diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index 4634de57073d..1b7d4fb5e17f 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -49,6 +49,7 @@
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.CatalogTests;
 import org.apache.iceberg.catalog.Namespace;
@@ -285,6 +286,50 @@ public void testBasicCatalog() throws Exception {
     catalog.dropTable(testTable);
   }
 
+  @Test
+  public void testCreateTableWithUniqueLocation() throws IOException {
+    try (JdbcCatalog jdbcCatalog =
+        initCatalog(
+            "unique_jdbc_catalog",
+            ImmutableMap.of(
+                String.format("table-default.%s", TableProperties.UNIQUE_LOCATION), "true"))) {
+      Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");
+      jdbcCatalog.createNamespace(testNamespace, Maps.newHashMap());
+      TableIdentifier tableIdent = TableIdentifier.of(testNamespace, "unique");
+      TableIdentifier tableRenamed = TableIdentifier.of(testNamespace, "renamed");
+
+      Table table = jdbcCatalog.createTable(tableIdent, SCHEMA, PartitionSpec.unpartitioned());
+      String currentLocation = table.location();
+
+      FileSystem fs = Util.getFs(new Path(currentLocation), conf);
+      assertThat(fs.isDirectory(new Path(currentLocation))).isTrue();
+      jdbcCatalog.renameTable(tableIdent, tableRenamed);
+
+      Assertions.assertThatThrownBy(
+              () ->
+                  jdbcCatalog.createTable(
+                      tableIdent,
+                      SCHEMA,
+                      PartitionSpec.unpartitioned(),
+                      currentLocation,
+                      ImmutableMap.of()))
+          .isInstanceOf(AlreadyExistsException.class)
+          .hasMessageStartingWith("Table location already in use");
+
+      Table recreated =
+          jdbcCatalog.createTable(
+              tableIdent,
+              SCHEMA,
+              PartitionSpec.unpartitioned(),
+              currentLocation,
+              ImmutableMap.of(TableProperties.UNIQUE_LOCATION, "false"));
+      assertThat(recreated.location()).isEqualTo(currentLocation);
+
+      jdbcCatalog.dropTable(tableRenamed);
+      jdbcCatalog.dropTable(tableIdent);
+    }
+  }
+
   @Test
   public void testCreateAndDropTableWithoutNamespace() throws Exception {
     TableIdentifier testTable = TableIdentifier.of("tbl");
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
index d4ac49868488..ffe2c2f2cba1 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
@@ -1162,6 +1162,43 @@ public void testDatabaseLocationWithSlashInWarehouseDir() {
     assertThat(database.getLocationUri()).isEqualTo("s3://bucket/database.db");
   }
 
+  @Test
+  public void testCreateTableWithUniqueLocation() {
+    Schema schema = getTestSchema();
+    TableIdentifier tableIdent = TableIdentifier.of(DB_NAME, "unique");
+    TableIdentifier tableRenamed = TableIdentifier.of(DB_NAME, "renamed");
+
+    ImmutableMap<String, String> catalogProps =
+        ImmutableMap.of(String.format("table-default.%s", TableProperties.UNIQUE_LOCATION), "true");
+    Catalog hiveCatalog =
+        CatalogUtil.loadCatalog(
+            HiveCatalog.class.getName(),
+            CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE,
+            catalogProps,
+            hiveConf);
+
+    try {
+      Table table = hiveCatalog.buildTable(tableIdent, schema).create();
+      String currentLocation = table.location();
+
+      catalog.renameTable(tableIdent, tableRenamed);
+
+      assertThatThrownBy(() -> hiveCatalog.buildTable(tableIdent, schema).create())
+          .isInstanceOf(AlreadyExistsException.class)
+          .hasMessageStartingWith("Table location already in use");
+
+      Table recreated =
+          hiveCatalog
+              .buildTable(tableIdent, schema)
+              .withProperty(TableProperties.UNIQUE_LOCATION, "false")
+              .create();
+      assertThat(recreated.location()).isEqualTo(currentLocation);
+    } finally {
+      hiveCatalog.dropTable(tableRenamed, true);
+      hiveCatalog.dropTable(tableIdent, true);
+    }
+  }
+
   @Test
   public void testRegisterTable() {
     TableIdentifier identifier = TableIdentifier.of(DB_NAME, "t1");