[#12988] fix(lance): reject nonempty Arrow input before table creation

lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java

@@ -168,11 +168,11 @@ public CreateTableResponse createTable(
     Preconditions.checkArgument(
         nsId.levels() == 3, "Expected at 3-level namespace but got: %s", nsId.levels());
 
-    // Parser column information.
+    // Reject unsupported record batches before any metadata or storage mutation.
     List<Column> columns = Lists.newArrayList();
     if (arrowStreamBody != null) {
       org.apache.arrow.vector.types.pojo.Schema schema =
-          ArrowUtils.parseArrowIpcStream(arrowStreamBody);
+          ArrowUtils.parseSchemaOnlyIpcStream(arrowStreamBody);
       columns = extractColumns(schema);
     }
 

lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java

@@ -57,16 +57,46 @@ public static byte[] generateIpcStream(Schema arrowSchema) throws IOException {
   }
 
   public static Schema parseArrowIpcStream(byte[] stream) {
+    return parseArrowIpcStream(stream, false);
+  }
+
+  /**
+   * Parses a schema-only Arrow IPC stream, rejecting record batches containing rows.
+   *
+   * @param stream the Arrow IPC stream
+   * @return the stream schema
+   * @throws UnsupportedOperationException if any record batch contains rows
+   * @throws IllegalArgumentException if the stream cannot be parsed
+   */
+  public static Schema parseSchemaOnlyIpcStream(byte[] stream) {
+    return parseArrowIpcStream(stream, true);
+  }
+
+  private static Schema parseArrowIpcStream(byte[] stream, boolean requireEmpty) {
     Schema schema;
+    boolean containsRows = false;
     try (BufferAllocator allocator = new RootAllocator();
         ByteArrayInputStream bais = new ByteArrayInputStream(stream);
         ArrowStreamReader reader = new ArrowStreamReader(bais, allocator)) {
       schema = reader.getVectorSchemaRoot().getSchema();
+      if (requireEmpty) {
+        while (reader.loadNextBatch()) {
+          if (reader.getVectorSchemaRoot().getRowCount() > 0) {
+            containsRows = true;
+            break;
+          }
+        }
+      }
     } catch (Exception e) {
       throw new IllegalArgumentException("Failed to parse Arrow IPC stream", e);
     }
 
     Preconditions.checkArgument(schema != null, "No schema found in Arrow IPC stream");
+    if (containsRows) {
+      throw new UnsupportedOperationException(
+          "CreateTable only supports schema-only Arrow streams; "
+              + "write records through a Lance client or engine after creation");
+    }
     return schema;
   }
 }

lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java

@@ -18,7 +18,13 @@
  */
 package org.apache.gravitino.lance.common.utils;
 
+import java.io.ByteArrayOutputStream;
 import java.util.Arrays;
+import java.util.List;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowStreamWriter;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.Schema;
@@ -39,4 +45,53 @@ public void testParseArrowIpcStream() throws Exception {
 
     Assertions.assertEquals(schema, parsedSchema);
   }
+  /** Verifies schema-only streams and zero-row batches remain supported. */
+  @Test
+  public void testSchemaOnlyStreams() throws Exception {
+    Schema expected = new Schema(List.of(Field.nullable("id", new ArrowType.Int(32, true))));
+    Assertions.assertEquals(expected, ArrowUtils.parseSchemaOnlyIpcStream(streamWithRows()));
+    Assertions.assertEquals(expected, ArrowUtils.parseSchemaOnlyIpcStream(streamWithRows(0, 0)));
+  }
+
+  /** Verifies that a non-empty batch is rejected, including after empty batches. */
+  @Test
+  public void testRejectRecordBatchesWithRows() throws Exception {
+    for (byte[] stream : List.of(streamWithRows(1), streamWithRows(0, 1))) {
+      UnsupportedOperationException exception =
+          Assertions.assertThrows(
+              UnsupportedOperationException.class,
+              () -> ArrowUtils.parseSchemaOnlyIpcStream(stream));
+      Assertions.assertTrue(exception.getMessage().contains("schema-only"));
+      // Existing callers of the general schema parser retain their previous behavior.
+      Assertions.assertEquals(1, ArrowUtils.parseArrowIpcStream(stream).getFields().size());
+    }
+  }
+
+  /** Verifies malformed input is reported as invalid rather than as unsupported data. */
+  @Test
+  public void testRejectMalformedSchemaOnlyStream() {
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> ArrowUtils.parseSchemaOnlyIpcStream(new byte[] {1, 2, 3}));
+  }
+
+  private byte[] streamWithRows(int... batches) throws Exception {
+    Schema schema = new Schema(List.of(Field.nullable("id", new ArrowType.Int(32, true))));
+    try (RootAllocator allocator = new RootAllocator();
+        VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator);
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        ArrowStreamWriter writer = new ArrowStreamWriter(root, null, output)) {
+      root.allocateNew();
+      writer.start();
+      for (int rows : batches) {
+        for (int i = 0; i < rows; i++) {
+          ((IntVector) root.getVector("id")).setSafe(i, i);
+        }
+        root.setRowCount(rows);
+        writer.writeBatch();
+      }
+      writer.end();
+      return output.toByteArray();
+    }
+  }
 }

lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java

@@ -21,6 +21,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.Charset;
@@ -34,6 +35,10 @@
 import java.util.Set;
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowStreamWriter;
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.commons.io.FileUtils;
@@ -52,6 +57,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.lance.Dataset;
 import org.lance.namespace.LanceNamespace;
 import org.lance.namespace.client.apache.ApiClient;
 import org.lance.namespace.client.apache.ApiException;
@@ -449,6 +455,45 @@ public void testNamespaceExists() {
     assertLanceErrorCode(exception, ErrorCode.NAMESPACE_NOT_FOUND);
   }
 
+  /** Verifies unsupported input cannot silently discard records or destroy an existing table. */
+  @Test
+  void testCreateRejectsNonEmptyArrowWithoutSideEffects() throws IOException {
+    catalog = createCatalog(CATALOG_NAME);
+    createSchema();
+    byte[] data = arrowStreamWithRecord();
+    for (String mode : List.of("create", "exist_ok")) {
+      String name = "nonempty_" + mode;
+      Path location = tempDir.resolve(name);
+      assertNonEmptyCreateRejected(
+          List.of(CATALOG_NAME, SCHEMA_NAME, name), location.toString(), data, mode);
+      Assertions.assertFalse(
+          catalog.asTableCatalog().tableExists(NameIdentifier.of(SCHEMA_NAME, name)));
+      Assertions.assertFalse(Files.exists(location));
+    }
+
+    String original = "nonempty_overwrite";
+    List<String> ids = List.of(CATALOG_NAME, SCHEMA_NAME, original);
+    String location = tempDir.resolve(original).toString();
+    try (VectorSchemaRoot root =
+        VectorSchemaRoot.of(
+            new IntVector("id", allocator), new VarCharVector("value", allocator))) {
+      createTable(
+          ids, location, Map.of(), ArrowUtils.generateIpcStream(root.getSchema()), "create");
+    }
+    assertNonEmptyCreateRejected(ids, location, data, "overwrite");
+    DescribeTableRequest describe = new DescribeTableRequest();
+    describe.setId(ids);
+    Assertions.assertEquals(
+        List.of("id", "value"),
+        ns.describeTable(describe).getSchema().getFields().stream()
+            .map(JsonArrowField::getName)
+            .toList());
+    try (Dataset dataset = Dataset.open().uri(location).build()) {
+      Assertions.assertEquals(0, dataset.countRows());
+      Assertions.assertEquals(2, dataset.getSchema().getFields().size());
+    }
+  }
+
   @Test
   void testCreateTable() throws IOException {
     catalog = createCatalog(CATALOG_NAME);
@@ -968,6 +1013,40 @@ void testDeclareTable() {
     Assertions.assertFalse(new File(anotherLocation).exists());
   }
 
+  private void assertNonEmptyCreateRejected(
+      List<String> ids, String location, byte[] data, String mode) {
+    ApiException error =
+        Assertions.assertThrows(
+            ApiException.class,
+            () ->
+                createTableApi()
+                    .createTable(
+                        String.join(DELIMITER, ids),
+                        data,
+                        DELIMITER,
+                        mode,
+                        null,
+                        null,
+                        Map.of(LanceConstants.LANCE_TABLE_LOCATION_HEADER, location)));
+    Assertions.assertEquals(406, error.getCode());
+  }
+
+  private byte[] arrowStreamWithRecord() throws IOException {
+    try (VectorSchemaRoot root = VectorSchemaRoot.of(new IntVector("id", allocator));
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        ArrowStreamWriter writer = new ArrowStreamWriter(root, null, output)) {
+      root.allocateNew();
+      root.setRowCount(0);
+      writer.start();
+      writer.writeBatch();
+      ((IntVector) root.getVector("id")).setSafe(0, 42);
+      root.setRowCount(1);
+      writer.writeBatch();
+      writer.end();
+      return output.toByteArray();
+    }
+  }
+
   private CreateTableResponse createTable(
       List<String> ids,
       String location,