Spark: SnapshotTableSparkAction add validation for non-overlapping source/dest table paths.

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SnapshotTableSparkAction.java

@@ -124,10 +124,12 @@ private SnapshotTable.Result doExecute() {
     StagedSparkTable stagedTable = stageDestTable();
     Table icebergTable = stagedTable.table();
 
-    // TODO: Check the dest table location does not overlap with the source table location
-
     boolean threw = true;
     try {
+      Preconditions.checkArgument(
+          !sourceTableLocation().equals(icebergTable.location()),
+          "The destination table location overlaps with the source table location.");
+
       LOG.info("Ensuring {} has a valid name mapping", destTableIdent());
       ensureNameMappingPresent(icebergTable);
 

spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestSnapshotTableAction.java

@@ -19,12 +19,16 @@
 package org.apache.iceberg.spark.actions;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 import java.io.IOException;
 import java.nio.file.Files;
+import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.spark.CatalogTestBase;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.TestTemplate;
@@ -65,4 +69,39 @@ public void testSnapshotWithParallelTasks() throws IOException {
         .execute();
     assertThat(snapshotThreadsIndex.get()).isEqualTo(2);
   }
+
+  @TestTemplate
+  public void testTableLocationOverlapThrowsException() throws IOException {
+    // Ensure the test runs only for non-Hadoop-based catalogs,
+    // because path-based tables cannot have a custom location set.
+    assumeTrue(
+        !catalogName.equals("testhadoop"), "Cannot set a custom location for a path-based table.");
+
+    String location = Files.createTempDirectory(temp, "junit").toFile().toString();
+
+    sql(
+        "CREATE TABLE %s (id bigint NOT NULL, data string) USING parquet LOCATION '%s'",
+        SOURCE_NAME, location);
+    sql("INSERT INTO TABLE %s VALUES (1, 'a')", SOURCE_NAME);
+    sql("INSERT INTO TABLE %s VALUES (2, 'b')", SOURCE_NAME);
+
+    // Define properties for the destination table, setting its location to the same path as the
+    // source table
+    Map<String, String> tableProperties = Maps.newHashMap();
+    tableProperties.put("location", "file:" + location);
+
+    // Test that an exception is thrown
+    // when the destination table location overlaps with the source table location
+    // Assert that the exception message matches the expected error message
+    assertThatThrownBy(
+            () ->
+                SparkActions.get()
+                    .snapshotTable(SOURCE_NAME)
+                    .as(tableName)
+                    .tableProperties(tableProperties)
+                    .execute())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining(
+            "The destination table location overlaps with the source table location.");
+  }
 }