Spark3.4: Throw a friendly exception if table is empty when creating branch

spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/CreateOrReplaceBranchExec.scala

@@ -41,7 +41,12 @@ case class CreateOrReplaceBranchExec(
   override protected def run(): Seq[InternalRow] = {
     catalog.loadTable(ident) match {
       case iceberg: SparkTable =>
-        val snapshotId = branchOptions.snapshotId.getOrElse(iceberg.table.currentSnapshot().snapshotId())
+        var snapshotId = branchOptions.snapshotId.getOrElse(-1L)
+        if (snapshotId == -1) {
+          val currentSnapshot = Option(iceberg.table().currentSnapshot()).getOrElse(throw new IllegalArgumentException(
+            s"Please specify an explicit snapshot as table: $iceberg"  + " has no latest main snapshot"))
+          snapshotId = currentSnapshot.snapshotId()
+        }
         val manageSnapshots = iceberg.table().manageSnapshots()
         if (!replace) {
           val ref = iceberg.table().refs().get(branch);

spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestBranchDDL.java

@@ -212,6 +212,17 @@ public void testCreateBranchUseCustomMaxRefAge() throws NoSuchTableException {
                 tableName, branchName, maxRefAge));
   }
 
+  @Test
+  public void testCreateBranchOnEmptyTable() throws NoSuchTableException {
+    createEmptyTable();
+    String branchName = "b1";
+    AssertHelpers.assertThrows(
+            "Illegal argument",
+            IllegalArgumentException.class,
+            "has no latest main snapshot",
+            () -> sql("ALTER TABLE %s CREATE BRANCH %s", tableName, branchName));
+  }
+
   @Test
   public void testDropBranch() throws NoSuchTableException {
     insertRows();
@@ -289,4 +300,8 @@ private Table insertRows() throws NoSuchTableException {
     df.writeTo(tableName).append();
     return validationCatalog.loadTable(tableIdent);
   }
+
+  private Table createEmptyTable() {
+    return validationCatalog.loadTable(tableIdent);
+  }
 }