From 87d1f978ff955c2395a8ccc97cda91b8b51c6586 Mon Sep 17 00:00:00 2001 From: yuqi Date: Tue, 8 Sep 2026 16:46:21 +0800 Subject: [PATCH 1/3] fix(lance): reject unsupported table data and preserve backend errors --- .../GravitinoLanceTableOperations.java | 4 +- .../lance/common/utils/ArrowUtils.java | 30 +++++ .../lance/common/utils/TestArrowUtils.java | 55 ++++++++ lance/lance-rest-server/build.gradle.kts | 1 + .../lance/service/LanceExceptionMapper.java | 12 +- ...etadataAuthorizationMethodInterceptor.java | 6 +- .../test/LanceNamespaceAuthorizationIT.java | 126 ++++++++++++++++++ .../test/LanceTableAuthorizationIT.java | 64 +++++++++ .../service/TestLanceExceptionMapper.java | 79 +++++++++++ .../rest/TestLanceNamespaceOperations.java | 29 ++-- 10 files changed, 386 insertions(+), 20 deletions(-) create mode 100644 lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java diff --git a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java index 08bd9bae3a9..49bd407e7ee 100644 --- a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/ops/gravitino/GravitinoLanceTableOperations.java +++ b/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 columns = Lists.newArrayList(); if (arrowStreamBody != null) { org.apache.arrow.vector.types.pojo.Schema schema = - ArrowUtils.parseArrowIpcStream(arrowStreamBody); + ArrowUtils.parseSchemaOnlyIpcStream(arrowStreamBody); columns = extractColumns(schema); } diff --git a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java index 5d8508ee459..d2213e5e748 100644 --- a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java +++ b/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; } } diff --git a/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java b/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java index 43f0bf6ec6f..8390e2a1bf0 100644 --- a/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java +++ b/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(); + } + } } diff --git a/lance/lance-rest-server/build.gradle.kts b/lance/lance-rest-server/build.gradle.kts index 176c4f16b7a..db4408e7850 100644 --- a/lance/lance-rest-server/build.gradle.kts +++ b/lance/lance-rest-server/build.gradle.kts @@ -193,6 +193,7 @@ tasks { val primaryBundleDir = lanceSparkBundleDirFor(primaryLanceSparkBundleVersion) doFirst { + systemProperty("lance.test.runtimeClasspath", sourceSets["main"].runtimeClasspath.asPath) val bundleJar = primaryBundleDir.get().asFile.listFiles()?.singleOrNull { it.extension == "jar" } ?: throw GradleException( diff --git a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java index 2078b75fe6f..45d4cb4f47f 100644 --- a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java +++ b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java @@ -23,8 +23,10 @@ import javax.ws.rs.core.Response; import javax.ws.rs.ext.ExceptionMapper; import javax.ws.rs.ext.Provider; +import org.apache.gravitino.exceptions.ForbiddenException; import org.apache.gravitino.exceptions.NoSuchTableException; import org.apache.gravitino.exceptions.NotFoundException; +import org.apache.gravitino.exceptions.UnauthorizedException; import org.lance.namespace.errors.ConcurrentModificationException; import org.lance.namespace.errors.InternalException; import org.lance.namespace.errors.InvalidInputException; @@ -66,7 +68,13 @@ public Response toResponse(Exception ex) { } private static LanceNamespaceException toLanceNamespaceException(String instance, Exception ex) { - if (ex instanceof NoSuchTableException) { + if (ex instanceof ForbiddenException) { + return new PermissionDeniedException(ex.getMessage(), "", instance); + + } else if (ex instanceof UnauthorizedException) { + return new UnauthenticatedException(ex.getMessage(), "", instance); + + } else if (ex instanceof NoSuchTableException) { return new TableNotFoundException(ex.getMessage(), getStackTrace(ex), instance); } else if (ex instanceof NotFoundException) { @@ -84,7 +92,7 @@ private static LanceNamespaceException toLanceNamespaceException(String instance } else { LOG.warn("Lance REST server unexpected exception:", ex); - return new InternalException(ex.getMessage(), getStackTrace(ex), instance); + return new InternalException("Internal server error", "", instance); } } diff --git a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java index 88b3d645ecd..3ec6e9c9037 100644 --- a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java +++ b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java @@ -18,8 +18,6 @@ */ package org.apache.gravitino.lance.service.authorization; -import static org.apache.commons.lang3.exception.ExceptionUtils.getStackTrace; - import java.lang.reflect.Method; import java.lang.reflect.Parameter; import java.util.HashMap; @@ -188,9 +186,7 @@ protected Object toErrorResponse(Method method, Object[] args, Throwable throwab String namespaceId = pathArgument(method.getParameters(), args, "id").orElse(""); Exception exception; if (throwable instanceof ForbiddenException) { - exception = - new PermissionDeniedException( - throwable.getMessage(), getStackTrace(throwable), namespaceId); + exception = new PermissionDeniedException(throwable.getMessage(), "", namespaceId); } else if (throwable instanceof Exception) { exception = (Exception) throwable; } else { diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java index 5226a20beb9..96215dc95e5 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java @@ -18,16 +18,21 @@ */ package org.apache.gravitino.lance.integration.test; +import java.io.Writer; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.ArrayList; import java.util.Base64; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Properties; +import java.util.concurrent.TimeUnit; import org.apache.gravitino.Configs; import org.apache.gravitino.auth.AuthConstants; import org.apache.gravitino.authorization.Privileges; @@ -35,14 +40,20 @@ import org.apache.gravitino.authorization.SecurableObjects; import org.apache.gravitino.client.GravitinoMetalake; import org.apache.gravitino.integration.test.util.BaseIT; +import org.apache.gravitino.integration.test.util.HttpUtils; +import org.apache.gravitino.lance.server.GravitinoLanceRESTServer; +import org.apache.gravitino.rest.RESTUtils; import org.apache.gravitino.server.web.ObjectMapperProvider; +import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.lance.namespace.model.CreateNamespaceRequest; import org.lance.namespace.model.DescribeNamespaceResponse; import org.lance.namespace.model.DropNamespaceRequest; +import org.lance.namespace.model.ErrorResponse; import org.lance.namespace.model.ListNamespacesResponse; /** Verifies namespace authorization and list filtering through auxiliary-mode Lance REST. */ @@ -181,6 +192,121 @@ public void testDropConcealsNamespacesTheCallerMayNotSee() throws Exception { assertStatus(403, drop(USER, HIDDEN_CATALOG, "skip", null)); } + /** Verifies standalone HTTP backend calls use service credentials rather than caller roles. */ + @Test + public void testStandaloneUsesBackendServiceIdentity(@TempDir Path directory) throws Exception { + int port = RESTUtils.findAvailablePort(10000, 11000); + String catalog = "lance_authz_standalone_catalog"; + String serviceUser = "lance_authz_standalone_user"; + GravitinoMetalake metalake = client.loadMetalake(getLanceRESTServerMetalakeName()); + metalake.addUser(serviceUser); + metalake.createRole( + "lance_authz_standalone_role", + new HashMap<>(), + List.of( + SecurableObjects.ofMetalake( + metalake.name(), + new ArrayList<>( + List.of(Privileges.UseCatalog.allow(), Privileges.CreateCatalog.allow()))))); + metalake.grantRolesToUser(List.of("lance_authz_standalone_role"), serviceUser); + Properties config = new Properties(); + config.setProperty(Configs.AUTHENTICATORS.getKey(), "simple"); + config.setProperty("gravitino.lance-rest.httpPort", String.valueOf(port)); + config.setProperty( + "gravitino.lance-rest.gravitino-uri", "http://localhost:" + getGravitinoServerPort()); + config.setProperty("gravitino.lance-rest.gravitino-metalake", getLanceRESTServerMetalakeName()); + config.setProperty("gravitino.lance-rest.gravitino-auth-type", "simple"); + config.setProperty("gravitino.lance-rest.gravitino-simple.user-name", serviceUser); + Path configFile = directory.resolve("standalone.conf"); + try (Writer writer = Files.newBufferedWriter(configFile)) { + config.store(writer, "Standalone Lance REST integration test"); + } + Path logFile = directory.resolve("standalone.log"); + // Use the production bootstrap in its own JVM: deploy mode has no local GravitinoEnv, + // while embedded mode must not share its backend environment with the standalone service. + ProcessBuilder builder = + new ProcessBuilder( + Path.of(System.getProperty("java.home"), "bin", "java").toString(), + "--add-opens=java.base/java.nio=ALL-UNNAMED", + "-cp", + System.getProperty("lance.test.runtimeClasspath"), + GravitinoLanceRESTServer.class.getName(), + configFile.toString()) + .redirectErrorStream(true) + .redirectOutput(logFile.toFile()); + builder.environment().put("GRAVITINO_TEST", "true"); + Process standalone = builder.start(); + try { + try { + Awaitility.await() + .atMost(60, TimeUnit.SECONDS) + .until( + () -> { + Assertions.assertTrue(standalone.isAlive(), "Standalone process exited"); + // Namespace initialization is lazy and occurs on the first metadata request. + return HttpUtils.isHttpServerUp( + "http://localhost:" + port + "/lance/health/live"); + }); + } catch (Exception | AssertionError e) { + throw new AssertionError("Standalone startup failed:\n" + Files.readString(logFile), e); + } + CreateNamespaceRequest body = new CreateNamespaceRequest(); + body.addIdItem(catalog); + HttpRequest request = + request(USER, "/v1/namespace/" + catalog + "/create") + .uri( + URI.create( + "http://localhost:" + + port + + "/lance/v1/namespace/" + + catalog + + "/create?delimiter=.")) + .setHeader(AuthConstants.X_GRAVITINO_ACTIVE_ROLES_HEADER, "NONE") + .POST( + HttpRequest.BodyPublishers.ofString( + ObjectMapperProvider.objectMapper().writeValueAsString(body))) + .build(); + // USER cannot create catalogs in auxiliary mode. The backend receives the service user's + // credentials and roles, despite USER selecting NONE on this incoming request. + assertStatus(200, httpClient.send(request, HttpResponse.BodyHandlers.ofString())); + Assertions.assertEquals( + serviceUser, + client + .loadMetalake(getLanceRESTServerMetalakeName()) + .loadCatalog(catalog) + .auditInfo() + .creator()); + // The backend service user cannot read this admin-owned schema. Even an incoming admin + // must receive the backend's 403, rather than 500 or the incoming caller's privileges. + HttpRequest deniedRequest = + request(ADMIN, "/v1/namespace/" + id(VISIBLE_CATALOG, VISIBLE_SCHEMA) + "/describe") + .uri( + URI.create( + "http://localhost:" + + port + + "/lance/v1/namespace/" + + id(VISIBLE_CATALOG, VISIBLE_SCHEMA) + + "/describe?delimiter=.")) + .POST(HttpRequest.BodyPublishers.ofString("{}")) + .build(); + HttpResponse deniedResponse = + httpClient.send(deniedRequest, HttpResponse.BodyHandlers.ofString()); + assertStatus(403, deniedResponse); + ErrorResponse error = + ObjectMapperProvider.objectMapper().readValue(deniedResponse.body(), ErrorResponse.class); + Assertions.assertEquals("", error.getDetail()); + Assertions.assertTrue(error.getError().contains(serviceUser), error.getError()); + assertStatus(200, drop(serviceUser, catalog, null, "cascade")); + } finally { + standalone.destroy(); + if (!standalone.waitFor(10, TimeUnit.SECONDS)) { + standalone.destroyForcibly(); + Assertions.assertTrue( + standalone.waitFor(10, TimeUnit.SECONDS), "Standalone process did not stop"); + } + } + } + private void grant(GravitinoMetalake metalake, String role, SecurableObject object) { metalake.createRole(role, new HashMap<>(), List.of(object)); metalake.grantRolesToUser(List.of(role), USER); diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java index c7722b05235..25a91d79a7f 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java @@ -18,16 +18,22 @@ */ package org.apache.gravitino.lance.integration.test; +import java.io.ByteArrayOutputStream; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.Base64; import java.util.HashMap; 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; @@ -49,6 +55,7 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.lance.Dataset; import org.lance.namespace.model.AlterTableDropColumnsRequest; import org.lance.namespace.model.CreateNamespaceRequest; import org.lance.namespace.model.DeclareTableRequest; @@ -343,6 +350,63 @@ public void testMutationConcealsTablesTheCallerMayNotSee() throws Exception { assertStatus(404, table(ADMIN, MISSING_TABLE, "deregister")); } + /** Verifies unsupported input cannot silently discard records or destroy an existing table. */ + @Test + public void testCreateRejectsNonEmptyArrowWithoutSideEffects() throws Exception { + byte[] data = arrowStreamWithRecord(); + for (String mode : List.of("create", "exist_ok")) { + String name = "nonempty_" + mode; + assertStatus(406, createWithData(PROBER, name, mode, data)); + assertStatus(404, table(ADMIN, WRITE_SCHEMA, name, "exists")); + Assertions.assertFalse(Files.exists(tempDir.resolve(name))); + } + + String original = "nonempty_overwrite"; + createTable(WRITE_SCHEMA, original); + assertStatus(406, createWithData(MUTATOR, original, "overwrite", data)); + Assertions.assertEquals( + List.of("id", "value"), + describe(ADMIN, WRITE_SCHEMA, original).getSchema().getFields().stream() + .map(field -> field.getName()) + .toList()); + try (Dataset dataset = Dataset.open().uri(location(original)).build()) { + Assertions.assertEquals(0, dataset.countRows()); + Assertions.assertEquals(2, dataset.getSchema().getFields().size()); + } + } + + private HttpResponse createWithData( + String user, String tableName, String mode, byte[] data) throws Exception { + HttpRequest req = + request( + user, + "/v1/table/" + id(CATALOG, WRITE_SCHEMA, tableName) + "/create", + "&mode=" + mode) + .setHeader("Content-Type", "application/vnd.apache.arrow.stream") + .setHeader(LanceConstants.LANCE_TABLE_LOCATION_HEADER, location(tableName)) + .POST(HttpRequest.BodyPublishers.ofByteArray(data)) + .build(); + return httpClient.send(req, HttpResponse.BodyHandlers.ofString()); + } + + private byte[] arrowStreamWithRecord() 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(); + 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 HttpResponse dropColumns(String user, String tableName, String column) throws Exception { return dropColumns(user, WRITE_SCHEMA, tableName, column); diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java new file mode 100644 index 00000000000..aea2c57943a --- /dev/null +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java @@ -0,0 +1,79 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.lance.service; + +import javax.ws.rs.core.Response; +import org.apache.gravitino.exceptions.ForbiddenException; +import org.apache.gravitino.exceptions.UnauthorizedException; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.lance.namespace.errors.InvalidInputException; +import org.lance.namespace.model.ErrorResponse; + +/** Verifies backend authentication failures retain their protocol status without stack traces. */ +public class TestLanceExceptionMapper { + + /** Verifies backend authorization failures use the Lance forbidden response. */ + @Test + public void testBackendForbidden() { + assertAuthenticationError(new ForbiddenException("Access denied"), 403); + } + + /** Verifies backend authentication failures use the Lance unauthenticated response. */ + @Test + public void testBackendUnauthorized() { + assertAuthenticationError(new UnauthorizedException("Invalid credentials"), 401); + } + + /** Verifies unexpected exceptions do not expose internal details in the response. */ + @Test + public void testInternalFailureDoesNotExposeException() { + try (Response response = + LanceExceptionMapper.toRESTResponse( + "catalog.schema.table", new RuntimeException("private-backend-detail"))) { + Assertions.assertEquals(500, response.getStatus()); + ErrorResponse error = (ErrorResponse) response.getEntity(); + Assertions.assertEquals("Internal server error", error.getError()); + Assertions.assertEquals("", error.getDetail()); + } + } + + /** Verifies intentional protocol validation details remain available to callers. */ + @Test + public void testProtocolValidationDetailsArePreserved() { + try (Response response = + LanceExceptionMapper.toRESTResponse( + "table", + new InvalidInputException("Invalid field", "field must be positive", "table"))) { + Assertions.assertEquals(400, response.getStatus()); + Assertions.assertEquals( + "field must be positive", ((ErrorResponse) response.getEntity()).getDetail()); + } + } + + private void assertAuthenticationError(Exception exception, int status) { + try (Response response = LanceExceptionMapper.toRESTResponse("catalog", exception)) { + Assertions.assertEquals(status, response.getStatus()); + ErrorResponse error = (ErrorResponse) response.getEntity(); + Assertions.assertEquals(exception.getMessage(), error.getError()); + Assertions.assertEquals("", error.getDetail()); + Assertions.assertEquals("catalog", error.getInstance()); + } + } +} diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java index a80c70a06b5..87069562bd1 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java @@ -208,10 +208,9 @@ public void testListNamespaces() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Test exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); Assertions.assertEquals("ns1.ns2", errorResp.getInstance()); - Assertions.assertNotNull(errorResp.getDetail()); - Assertions.assertTrue(errorResp.getDetail().contains("Test exception")); // root endpoint should use explicit root identifier instead of delimiter in error instance resp = @@ -262,7 +261,8 @@ public void testDescribeNamespace() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Test exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -321,7 +321,8 @@ public void testCreateNamespace() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Test exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -393,7 +394,8 @@ public void testDropNamespace() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Test exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -457,7 +459,8 @@ void testCreateTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Runtime exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -513,7 +516,8 @@ void testRegisterTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Runtime exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -620,7 +624,8 @@ void testDeregisterTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Runtime exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -677,7 +682,8 @@ void testDescribeTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Runtime exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } @Test @@ -999,6 +1005,7 @@ void testDeclareTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Runtime exception", errorResp.getError()); + Assertions.assertEquals("Internal server error", errorResp.getError()); + Assertions.assertEquals("", errorResp.getDetail()); } } From 01a3f7b79bbdabac4b26195b1ebf9f04350b4746 Mon Sep 17 00:00:00 2001 From: yuqi Date: Tue, 8 Sep 2026 17:12:52 +0800 Subject: [PATCH 2/3] fix(lance): limit Arrow fix to input validation and independent tests --- lance/lance-rest-server/build.gradle.kts | 1 - .../lance/service/LanceExceptionMapper.java | 12 +- ...etadataAuthorizationMethodInterceptor.java | 6 +- .../test/LanceNamespaceAuthorizationIT.java | 126 ------------------ .../integration/test/LanceRESTServiceIT.java | 79 +++++++++++ .../test/LanceTableAuthorizationIT.java | 64 --------- .../service/TestLanceExceptionMapper.java | 79 ----------- .../rest/TestLanceNamespaceOperations.java | 29 ++-- 8 files changed, 97 insertions(+), 299 deletions(-) delete mode 100644 lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java diff --git a/lance/lance-rest-server/build.gradle.kts b/lance/lance-rest-server/build.gradle.kts index db4408e7850..176c4f16b7a 100644 --- a/lance/lance-rest-server/build.gradle.kts +++ b/lance/lance-rest-server/build.gradle.kts @@ -193,7 +193,6 @@ tasks { val primaryBundleDir = lanceSparkBundleDirFor(primaryLanceSparkBundleVersion) doFirst { - systemProperty("lance.test.runtimeClasspath", sourceSets["main"].runtimeClasspath.asPath) val bundleJar = primaryBundleDir.get().asFile.listFiles()?.singleOrNull { it.extension == "jar" } ?: throw GradleException( diff --git a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java index 45d4cb4f47f..2078b75fe6f 100644 --- a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java +++ b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/LanceExceptionMapper.java @@ -23,10 +23,8 @@ import javax.ws.rs.core.Response; import javax.ws.rs.ext.ExceptionMapper; import javax.ws.rs.ext.Provider; -import org.apache.gravitino.exceptions.ForbiddenException; import org.apache.gravitino.exceptions.NoSuchTableException; import org.apache.gravitino.exceptions.NotFoundException; -import org.apache.gravitino.exceptions.UnauthorizedException; import org.lance.namespace.errors.ConcurrentModificationException; import org.lance.namespace.errors.InternalException; import org.lance.namespace.errors.InvalidInputException; @@ -68,13 +66,7 @@ public Response toResponse(Exception ex) { } private static LanceNamespaceException toLanceNamespaceException(String instance, Exception ex) { - if (ex instanceof ForbiddenException) { - return new PermissionDeniedException(ex.getMessage(), "", instance); - - } else if (ex instanceof UnauthorizedException) { - return new UnauthenticatedException(ex.getMessage(), "", instance); - - } else if (ex instanceof NoSuchTableException) { + if (ex instanceof NoSuchTableException) { return new TableNotFoundException(ex.getMessage(), getStackTrace(ex), instance); } else if (ex instanceof NotFoundException) { @@ -92,7 +84,7 @@ private static LanceNamespaceException toLanceNamespaceException(String instance } else { LOG.warn("Lance REST server unexpected exception:", ex); - return new InternalException("Internal server error", "", instance); + return new InternalException(ex.getMessage(), getStackTrace(ex), instance); } } diff --git a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java index 3ec6e9c9037..88b3d645ecd 100644 --- a/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java +++ b/lance/lance-rest-server/src/main/java/org/apache/gravitino/lance/service/authorization/LanceMetadataAuthorizationMethodInterceptor.java @@ -18,6 +18,8 @@ */ package org.apache.gravitino.lance.service.authorization; +import static org.apache.commons.lang3.exception.ExceptionUtils.getStackTrace; + import java.lang.reflect.Method; import java.lang.reflect.Parameter; import java.util.HashMap; @@ -186,7 +188,9 @@ protected Object toErrorResponse(Method method, Object[] args, Throwable throwab String namespaceId = pathArgument(method.getParameters(), args, "id").orElse(""); Exception exception; if (throwable instanceof ForbiddenException) { - exception = new PermissionDeniedException(throwable.getMessage(), "", namespaceId); + exception = + new PermissionDeniedException( + throwable.getMessage(), getStackTrace(throwable), namespaceId); } else if (throwable instanceof Exception) { exception = (Exception) throwable; } else { diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java index 96215dc95e5..5226a20beb9 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceNamespaceAuthorizationIT.java @@ -18,21 +18,16 @@ */ package org.apache.gravitino.lance.integration.test; -import java.io.Writer; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.nio.file.Path; import java.util.ArrayList; import java.util.Base64; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; -import java.util.concurrent.TimeUnit; import org.apache.gravitino.Configs; import org.apache.gravitino.auth.AuthConstants; import org.apache.gravitino.authorization.Privileges; @@ -40,20 +35,14 @@ import org.apache.gravitino.authorization.SecurableObjects; import org.apache.gravitino.client.GravitinoMetalake; import org.apache.gravitino.integration.test.util.BaseIT; -import org.apache.gravitino.integration.test.util.HttpUtils; -import org.apache.gravitino.lance.server.GravitinoLanceRESTServer; -import org.apache.gravitino.rest.RESTUtils; import org.apache.gravitino.server.web.ObjectMapperProvider; -import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.io.TempDir; import org.lance.namespace.model.CreateNamespaceRequest; import org.lance.namespace.model.DescribeNamespaceResponse; import org.lance.namespace.model.DropNamespaceRequest; -import org.lance.namespace.model.ErrorResponse; import org.lance.namespace.model.ListNamespacesResponse; /** Verifies namespace authorization and list filtering through auxiliary-mode Lance REST. */ @@ -192,121 +181,6 @@ public void testDropConcealsNamespacesTheCallerMayNotSee() throws Exception { assertStatus(403, drop(USER, HIDDEN_CATALOG, "skip", null)); } - /** Verifies standalone HTTP backend calls use service credentials rather than caller roles. */ - @Test - public void testStandaloneUsesBackendServiceIdentity(@TempDir Path directory) throws Exception { - int port = RESTUtils.findAvailablePort(10000, 11000); - String catalog = "lance_authz_standalone_catalog"; - String serviceUser = "lance_authz_standalone_user"; - GravitinoMetalake metalake = client.loadMetalake(getLanceRESTServerMetalakeName()); - metalake.addUser(serviceUser); - metalake.createRole( - "lance_authz_standalone_role", - new HashMap<>(), - List.of( - SecurableObjects.ofMetalake( - metalake.name(), - new ArrayList<>( - List.of(Privileges.UseCatalog.allow(), Privileges.CreateCatalog.allow()))))); - metalake.grantRolesToUser(List.of("lance_authz_standalone_role"), serviceUser); - Properties config = new Properties(); - config.setProperty(Configs.AUTHENTICATORS.getKey(), "simple"); - config.setProperty("gravitino.lance-rest.httpPort", String.valueOf(port)); - config.setProperty( - "gravitino.lance-rest.gravitino-uri", "http://localhost:" + getGravitinoServerPort()); - config.setProperty("gravitino.lance-rest.gravitino-metalake", getLanceRESTServerMetalakeName()); - config.setProperty("gravitino.lance-rest.gravitino-auth-type", "simple"); - config.setProperty("gravitino.lance-rest.gravitino-simple.user-name", serviceUser); - Path configFile = directory.resolve("standalone.conf"); - try (Writer writer = Files.newBufferedWriter(configFile)) { - config.store(writer, "Standalone Lance REST integration test"); - } - Path logFile = directory.resolve("standalone.log"); - // Use the production bootstrap in its own JVM: deploy mode has no local GravitinoEnv, - // while embedded mode must not share its backend environment with the standalone service. - ProcessBuilder builder = - new ProcessBuilder( - Path.of(System.getProperty("java.home"), "bin", "java").toString(), - "--add-opens=java.base/java.nio=ALL-UNNAMED", - "-cp", - System.getProperty("lance.test.runtimeClasspath"), - GravitinoLanceRESTServer.class.getName(), - configFile.toString()) - .redirectErrorStream(true) - .redirectOutput(logFile.toFile()); - builder.environment().put("GRAVITINO_TEST", "true"); - Process standalone = builder.start(); - try { - try { - Awaitility.await() - .atMost(60, TimeUnit.SECONDS) - .until( - () -> { - Assertions.assertTrue(standalone.isAlive(), "Standalone process exited"); - // Namespace initialization is lazy and occurs on the first metadata request. - return HttpUtils.isHttpServerUp( - "http://localhost:" + port + "/lance/health/live"); - }); - } catch (Exception | AssertionError e) { - throw new AssertionError("Standalone startup failed:\n" + Files.readString(logFile), e); - } - CreateNamespaceRequest body = new CreateNamespaceRequest(); - body.addIdItem(catalog); - HttpRequest request = - request(USER, "/v1/namespace/" + catalog + "/create") - .uri( - URI.create( - "http://localhost:" - + port - + "/lance/v1/namespace/" - + catalog - + "/create?delimiter=.")) - .setHeader(AuthConstants.X_GRAVITINO_ACTIVE_ROLES_HEADER, "NONE") - .POST( - HttpRequest.BodyPublishers.ofString( - ObjectMapperProvider.objectMapper().writeValueAsString(body))) - .build(); - // USER cannot create catalogs in auxiliary mode. The backend receives the service user's - // credentials and roles, despite USER selecting NONE on this incoming request. - assertStatus(200, httpClient.send(request, HttpResponse.BodyHandlers.ofString())); - Assertions.assertEquals( - serviceUser, - client - .loadMetalake(getLanceRESTServerMetalakeName()) - .loadCatalog(catalog) - .auditInfo() - .creator()); - // The backend service user cannot read this admin-owned schema. Even an incoming admin - // must receive the backend's 403, rather than 500 or the incoming caller's privileges. - HttpRequest deniedRequest = - request(ADMIN, "/v1/namespace/" + id(VISIBLE_CATALOG, VISIBLE_SCHEMA) + "/describe") - .uri( - URI.create( - "http://localhost:" - + port - + "/lance/v1/namespace/" - + id(VISIBLE_CATALOG, VISIBLE_SCHEMA) - + "/describe?delimiter=.")) - .POST(HttpRequest.BodyPublishers.ofString("{}")) - .build(); - HttpResponse deniedResponse = - httpClient.send(deniedRequest, HttpResponse.BodyHandlers.ofString()); - assertStatus(403, deniedResponse); - ErrorResponse error = - ObjectMapperProvider.objectMapper().readValue(deniedResponse.body(), ErrorResponse.class); - Assertions.assertEquals("", error.getDetail()); - Assertions.assertTrue(error.getError().contains(serviceUser), error.getError()); - assertStatus(200, drop(serviceUser, catalog, null, "cascade")); - } finally { - standalone.destroy(); - if (!standalone.waitFor(10, TimeUnit.SECONDS)) { - standalone.destroyForcibly(); - Assertions.assertTrue( - standalone.waitFor(10, TimeUnit.SECONDS), "Standalone process did not stop"); - } - } - } - private void grant(GravitinoMetalake metalake, String role, SecurableObject object) { metalake.createRole(role, new HashMap<>(), List.of(object)); metalake.grantRolesToUser(List.of(role), USER); diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java index dcaa97c7f24..efdfe58dd30 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java +++ b/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 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 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 ids, String location, diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java index 25a91d79a7f..c7722b05235 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceTableAuthorizationIT.java @@ -18,22 +18,16 @@ */ package org.apache.gravitino.lance.integration.test; -import java.io.ByteArrayOutputStream; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; -import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.Base64; import java.util.HashMap; 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; @@ -55,7 +49,6 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; -import org.lance.Dataset; import org.lance.namespace.model.AlterTableDropColumnsRequest; import org.lance.namespace.model.CreateNamespaceRequest; import org.lance.namespace.model.DeclareTableRequest; @@ -350,63 +343,6 @@ public void testMutationConcealsTablesTheCallerMayNotSee() throws Exception { assertStatus(404, table(ADMIN, MISSING_TABLE, "deregister")); } - /** Verifies unsupported input cannot silently discard records or destroy an existing table. */ - @Test - public void testCreateRejectsNonEmptyArrowWithoutSideEffects() throws Exception { - byte[] data = arrowStreamWithRecord(); - for (String mode : List.of("create", "exist_ok")) { - String name = "nonempty_" + mode; - assertStatus(406, createWithData(PROBER, name, mode, data)); - assertStatus(404, table(ADMIN, WRITE_SCHEMA, name, "exists")); - Assertions.assertFalse(Files.exists(tempDir.resolve(name))); - } - - String original = "nonempty_overwrite"; - createTable(WRITE_SCHEMA, original); - assertStatus(406, createWithData(MUTATOR, original, "overwrite", data)); - Assertions.assertEquals( - List.of("id", "value"), - describe(ADMIN, WRITE_SCHEMA, original).getSchema().getFields().stream() - .map(field -> field.getName()) - .toList()); - try (Dataset dataset = Dataset.open().uri(location(original)).build()) { - Assertions.assertEquals(0, dataset.countRows()); - Assertions.assertEquals(2, dataset.getSchema().getFields().size()); - } - } - - private HttpResponse createWithData( - String user, String tableName, String mode, byte[] data) throws Exception { - HttpRequest req = - request( - user, - "/v1/table/" + id(CATALOG, WRITE_SCHEMA, tableName) + "/create", - "&mode=" + mode) - .setHeader("Content-Type", "application/vnd.apache.arrow.stream") - .setHeader(LanceConstants.LANCE_TABLE_LOCATION_HEADER, location(tableName)) - .POST(HttpRequest.BodyPublishers.ofByteArray(data)) - .build(); - return httpClient.send(req, HttpResponse.BodyHandlers.ofString()); - } - - private byte[] arrowStreamWithRecord() 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(); - 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 HttpResponse dropColumns(String user, String tableName, String column) throws Exception { return dropColumns(user, WRITE_SCHEMA, tableName, column); diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java deleted file mode 100644 index aea2c57943a..00000000000 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/TestLanceExceptionMapper.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.gravitino.lance.service; - -import javax.ws.rs.core.Response; -import org.apache.gravitino.exceptions.ForbiddenException; -import org.apache.gravitino.exceptions.UnauthorizedException; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; -import org.lance.namespace.errors.InvalidInputException; -import org.lance.namespace.model.ErrorResponse; - -/** Verifies backend authentication failures retain their protocol status without stack traces. */ -public class TestLanceExceptionMapper { - - /** Verifies backend authorization failures use the Lance forbidden response. */ - @Test - public void testBackendForbidden() { - assertAuthenticationError(new ForbiddenException("Access denied"), 403); - } - - /** Verifies backend authentication failures use the Lance unauthenticated response. */ - @Test - public void testBackendUnauthorized() { - assertAuthenticationError(new UnauthorizedException("Invalid credentials"), 401); - } - - /** Verifies unexpected exceptions do not expose internal details in the response. */ - @Test - public void testInternalFailureDoesNotExposeException() { - try (Response response = - LanceExceptionMapper.toRESTResponse( - "catalog.schema.table", new RuntimeException("private-backend-detail"))) { - Assertions.assertEquals(500, response.getStatus()); - ErrorResponse error = (ErrorResponse) response.getEntity(); - Assertions.assertEquals("Internal server error", error.getError()); - Assertions.assertEquals("", error.getDetail()); - } - } - - /** Verifies intentional protocol validation details remain available to callers. */ - @Test - public void testProtocolValidationDetailsArePreserved() { - try (Response response = - LanceExceptionMapper.toRESTResponse( - "table", - new InvalidInputException("Invalid field", "field must be positive", "table"))) { - Assertions.assertEquals(400, response.getStatus()); - Assertions.assertEquals( - "field must be positive", ((ErrorResponse) response.getEntity()).getDetail()); - } - } - - private void assertAuthenticationError(Exception exception, int status) { - try (Response response = LanceExceptionMapper.toRESTResponse("catalog", exception)) { - Assertions.assertEquals(status, response.getStatus()); - ErrorResponse error = (ErrorResponse) response.getEntity(); - Assertions.assertEquals(exception.getMessage(), error.getError()); - Assertions.assertEquals("", error.getDetail()); - Assertions.assertEquals("catalog", error.getInstance()); - } - } -} diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java index 87069562bd1..a80c70a06b5 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/service/rest/TestLanceNamespaceOperations.java @@ -208,9 +208,10 @@ public void testListNamespaces() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Test exception", errorResp.getError()); Assertions.assertEquals("ns1.ns2", errorResp.getInstance()); + Assertions.assertNotNull(errorResp.getDetail()); + Assertions.assertTrue(errorResp.getDetail().contains("Test exception")); // root endpoint should use explicit root identifier instead of delimiter in error instance resp = @@ -261,8 +262,7 @@ public void testDescribeNamespace() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Test exception", errorResp.getError()); } @Test @@ -321,8 +321,7 @@ public void testCreateNamespace() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Test exception", errorResp.getError()); } @Test @@ -394,8 +393,7 @@ public void testDropNamespace() { ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); Assertions.assertEquals(18, errorResp.getCode()); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Test exception", errorResp.getError()); } @Test @@ -459,8 +457,7 @@ void testCreateTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Runtime exception", errorResp.getError()); } @Test @@ -516,8 +513,7 @@ void testRegisterTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Runtime exception", errorResp.getError()); } @Test @@ -624,8 +620,7 @@ void testDeregisterTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Runtime exception", errorResp.getError()); } @Test @@ -682,8 +677,7 @@ void testDescribeTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Runtime exception", errorResp.getError()); } @Test @@ -1005,7 +999,6 @@ void testDeclareTable() { Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), resp.getStatus()); Assertions.assertEquals(MediaType.APPLICATION_JSON_TYPE, resp.getMediaType()); ErrorResponse errorResp = resp.readEntity(ErrorResponse.class); - Assertions.assertEquals("Internal server error", errorResp.getError()); - Assertions.assertEquals("", errorResp.getDetail()); + Assertions.assertEquals("Runtime exception", errorResp.getError()); } } From 8dee02f4c885161fa736a4c53b24b54957e7e12c Mon Sep 17 00:00:00 2001 From: yuqi Date: Tue, 8 Sep 2026 20:00:12 +0800 Subject: [PATCH 3/3] fix(lance): inspect Arrow batch metadata without decoding rows --- .../lance/common/utils/ArrowUtils.java | 36 +++++-- .../lance/common/utils/TestArrowUtils.java | 98 +++++++++++++++++++ .../integration/test/LanceRESTServiceIT.java | 4 +- 3 files changed, 130 insertions(+), 8 deletions(-) diff --git a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java index d2213e5e748..b72ed08f7e9 100644 --- a/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java +++ b/lance/lance-common/src/main/java/org/apache/gravitino/lance/common/utils/ArrowUtils.java @@ -23,11 +23,17 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.nio.channels.Channels; +import org.apache.arrow.flatbuf.Message; +import org.apache.arrow.flatbuf.MessageHeader; +import org.apache.arrow.flatbuf.RecordBatch; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.memory.RootAllocator; import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.ipc.ArrowStreamReader; import org.apache.arrow.vector.ipc.ArrowStreamWriter; +import org.apache.arrow.vector.ipc.ReadChannel; +import org.apache.arrow.vector.ipc.message.MessageMetadataResult; +import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.pojo.Schema; public class ArrowUtils { @@ -80,12 +86,7 @@ private static Schema parseArrowIpcStream(byte[] stream, boolean requireEmpty) { ArrowStreamReader reader = new ArrowStreamReader(bais, allocator)) { schema = reader.getVectorSchemaRoot().getSchema(); if (requireEmpty) { - while (reader.loadNextBatch()) { - if (reader.getVectorSchemaRoot().getRowCount() > 0) { - containsRows = true; - break; - } - } + containsRows = containsRecordBatchRows(bais); } } catch (Exception e) { throw new IllegalArgumentException("Failed to parse Arrow IPC stream", e); @@ -99,4 +100,27 @@ private static Schema parseArrowIpcStream(byte[] stream, boolean requireEmpty) { } return schema; } + + private static boolean containsRecordBatchRows(ByteArrayInputStream input) throws IOException { + // The schema reader has consumed the schema message. Inspect only subsequent message headers; + // skipping bodies avoids allocating or decoding vectors, including dictionary values. + try (ReadChannel channel = new ReadChannel(Channels.newChannel(input))) { + MessageMetadataResult metadata; + while ((metadata = MessageSerializer.readMessage(channel)) != null) { + Message message = metadata.getMessage(); + if (message.headerType() == MessageHeader.RecordBatch) { + RecordBatch batch = (RecordBatch) message.header(new RecordBatch()); + Preconditions.checkArgument(batch.length() >= 0, "Invalid Arrow record batch row count"); + if (batch.length() > 0) { + return true; + } + } else if (message.headerType() != MessageHeader.DictionaryBatch) { + throw new IOException("Unexpected Arrow message type: " + message.headerType()); + } + Preconditions.checkArgument(message.bodyLength() >= 0, "Invalid Arrow message body length"); + input.skipNBytes(message.bodyLength()); + } + return false; + } + } } diff --git a/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java b/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java index 8390e2a1bf0..7f1a7ec4828 100644 --- a/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java +++ b/lance/lance-common/src/test/java/org/apache/gravitino/lance/common/utils/TestArrowUtils.java @@ -18,15 +18,25 @@ */ package org.apache.gravitino.lance.common.utils; +import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.nio.channels.Channels; 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.VarCharVector; import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.dictionary.Dictionary; +import org.apache.arrow.vector.dictionary.DictionaryProvider.MapDictionaryProvider; import org.apache.arrow.vector.ipc.ArrowStreamWriter; +import org.apache.arrow.vector.ipc.ReadChannel; +import org.apache.arrow.vector.ipc.message.MessageMetadataResult; +import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.DictionaryEncoding; import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; import org.apache.arrow.vector.types.pojo.Schema; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -75,6 +85,94 @@ public void testRejectMalformedSchemaOnlyStream() { () -> ArrowUtils.parseSchemaOnlyIpcStream(new byte[] {1, 2, 3})); } + /** Verifies non-empty batches are rejected from metadata without decoding their bodies. */ + @Test + public void testRejectRowsBeforeReadingBatchBody() throws Exception { + byte[] stream = streamWithRows(1); + try (ReadChannel channel = + new ReadChannel(Channels.newChannel(new ByteArrayInputStream(stream)))) { + MessageSerializer.deserializeSchema(channel); + MessageMetadataResult batch = MessageSerializer.readMessage(channel); + Assertions.assertTrue(batch.getMessageBodyLength() > 0); + byte[] headersOnly = Arrays.copyOf(stream, (int) channel.bytesRead()); + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> ArrowUtils.parseSchemaOnlyIpcStream(headersOnly)); + } + } + + /** Verifies dictionary values are skipped and do not count as table rows. */ + @Test + public void testDictionaryBatches() throws Exception { + for (int rows : new int[] {0, 1}) { + byte[] stream = dictionaryStreamWithRows(rows); + if (rows == 0) { + Assertions.assertEquals( + ArrowUtils.parseArrowIpcStream(stream), ArrowUtils.parseSchemaOnlyIpcStream(stream)); + } else { + Assertions.assertThrows( + UnsupportedOperationException.class, () -> ArrowUtils.parseSchemaOnlyIpcStream(stream)); + } + } + } + + /** Verifies skipping a truncated dictionary body still reports malformed input. */ + @Test + public void testRejectTruncatedDictionaryBody() throws Exception { + byte[] stream = dictionaryStreamWithRows(0); + try (ReadChannel channel = + new ReadChannel(Channels.newChannel(new ByteArrayInputStream(stream)))) { + MessageSerializer.deserializeSchema(channel); + MessageMetadataResult dictionary = MessageSerializer.readMessage(channel); + Assertions.assertTrue(dictionary.getMessageBodyLength() > 0); + byte[] truncated = Arrays.copyOf(stream, (int) channel.bytesRead() + 1); + Assertions.assertThrows( + IllegalArgumentException.class, () -> ArrowUtils.parseSchemaOnlyIpcStream(truncated)); + } + } + + /** Verifies a schema message cannot appear where a record batch is expected. */ + @Test + public void testRejectUnexpectedMessage() throws Exception { + byte[] stream = streamWithRows(); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (ReadChannel channel = + new ReadChannel(Channels.newChannel(new ByteArrayInputStream(stream)))) { + MessageSerializer.deserializeSchema(channel); + output.write(stream, 0, (int) channel.bytesRead()); + output.write(stream); + } + Assertions.assertThrows( + IllegalArgumentException.class, + () -> ArrowUtils.parseSchemaOnlyIpcStream(output.toByteArray())); + } + + private byte[] dictionaryStreamWithRows(int rows) throws Exception { + DictionaryEncoding encoding = new DictionaryEncoding(0, false, new ArrowType.Int(32, true)); + Schema schema = + new Schema( + List.of(new Field("id", new FieldType(true, encoding.getIndexType(), encoding), null))); + try (RootAllocator allocator = new RootAllocator(); + VarCharVector values = new VarCharVector("values", allocator); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator); + ByteArrayOutputStream output = new ByteArrayOutputStream()) { + values.allocateNew(); + values.setSafe(0, new byte[] {42}); + values.setValueCount(1); + MapDictionaryProvider dictionaries = + new MapDictionaryProvider(new Dictionary(values, encoding)); + try (ArrowStreamWriter writer = new ArrowStreamWriter(root, dictionaries, output)) { + root.allocateNew(); + ((IntVector) root.getVector("id")).setSafe(0, 0); + root.setRowCount(rows); + writer.start(); + writer.writeBatch(); + writer.end(); + } + return output.toByteArray(); + } + } + 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(); diff --git a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java index efdfe58dd30..b4762fedf0d 100644 --- a/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java +++ b/lance/lance-rest-server/src/test/java/org/apache/gravitino/lance/integration/test/LanceRESTServiceIT.java @@ -460,7 +460,7 @@ public void testNamespaceExists() { void testCreateRejectsNonEmptyArrowWithoutSideEffects() throws IOException { catalog = createCatalog(CATALOG_NAME); createSchema(); - byte[] data = arrowStreamWithRecord(); + byte[] data = arrowStreamWithEmptyThenNonEmptyBatch(); for (String mode : List.of("create", "exist_ok")) { String name = "nonempty_" + mode; Path location = tempDir.resolve(name); @@ -1031,7 +1031,7 @@ private void assertNonEmptyCreateRejected( Assertions.assertEquals(406, error.getCode()); } - private byte[] arrowStreamWithRecord() throws IOException { + private byte[] arrowStreamWithEmptyThenNonEmptyBatch() throws IOException { try (VectorSchemaRoot root = VectorSchemaRoot.of(new IntVector("id", allocator)); ByteArrayOutputStream output = new ByteArrayOutputStream(); ArrowStreamWriter writer = new ArrowStreamWriter(root, null, output)) {