test: cover managed backfill commits on rest-dir #782
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -12,29 +12,195 @@ | |
| * limitations under the License. | ||
| */ | ||
|
|
||
| import com.sun.net.httpserver.HttpExchange; | ||
| import com.sun.net.httpserver.HttpServer; | ||
| import java.io.IOException; | ||
| import java.net.InetSocketAddress; | ||
| 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.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
| import org.lance.namespace.RestAdapter; | ||
|
|
||
| public final class LanceRestDirNamespaceServer { | ||
| private static final String CREATE_TABLE_VERSION_COUNT_PATH = | ||
| "/__lance_test/create_table_version_count"; | ||
|
|
||
| private LanceRestDirNamespaceServer() {} | ||
|
|
||
| public static void main(String[] args) throws Exception { | ||
| String root = args.length > 0 ? args[0] : "/home/lance/rest-data"; | ||
| String host = args.length > 1 ? args[1] : "127.0.0.1"; | ||
| int port = args.length > 2 ? Integer.parseInt(args[2]) : 10024; | ||
| boolean managedVersioning = args.length > 3 && Boolean.parseBoolean(args[3]); | ||
|
|
||
| Map<String, String> backendConfig = new HashMap<>(); | ||
| backendConfig.put("root", root); | ||
| if (managedVersioning) { | ||
| // DirectoryNamespace uses manifest storage to track versions managed by the namespace. | ||
| backendConfig.put("manifest_enabled", "true"); | ||
| backendConfig.put("table_version_tracking_enabled", "true"); | ||
|
There was a problem hiding this comment. This configuration advertises managed versioning, but the directory backend still accepts and exposes direct dataset commits, so the existing result-only tests are not sensitive to whether the backfill used the namespace commit API. Against the exact Please add an observable namespace-commit assertion—for example, a test-server counter/log or equivalent signal that advances only through ReproducerRun in JShell with the Maven test classpath and the JNI temp directory on an executable filesystem: backendConfig.put("root", testRoot);
backendConfig.put("manifest_enabled", "true");
backendConfig.put("table_version_tracking_enabled", "true");
var adapter = new RestAdapter("dir", backendConfig, "127.0.0.1", 0);
adapter.start();
clientConfig.put("uri", "http://127.0.0.1:" + adapter.getPort());
var client = new RestNamespace();
client.initialize(clientConfig, allocator);
client.createNamespace(new CreateNamespaceRequest().id(List.of("default")).mode("create"));
var tableId = List.of("default", "gate_table");
var declared = client.declareTable(new DeclareTableRequest().id(tableId));
var created = Dataset.create(allocator, declared.getLocation(), schema,
new WriteParams.Builder().withMode(WriteParams.WriteMode.CREATE).build());
var updateMap = UpdateMap.builder()
.updates(Map.of("registered", "true")).replace(false).build();
var txn = new Transaction.Builder().readVersion(created.version())
.operation(UpdateConfig.builder().configUpdates(updateMap).build()).build();
var registered = new CommitBuilder(created)
.namespaceClient(client).tableId(tableId)
.namespaceClientManagedVersioning(true).execute(txn);
registered.addColumns(List.of(extra)); // deliberate direct commit
var opened = Dataset.open().allocator(allocator)
.namespaceClient(client).tableId(tableId).build();
System.out.println(opened.version());
System.out.println(opened.getSchema().getFields().stream()
.map(Field::getName).toList());Observed: There was a problem hiding this comment. Fixed in 75c318c: the managed proxy now counts successful create-table-version requests, and the revised tests verify that a namespace commit increments the counter while a deliberate direct commit does not.\n\n There was a problem hiding this comment. Fixed in 75c318c: the managed proxy now counts successful create-table-version requests, and the revised tests verify that a namespace commit increments the counter while a deliberate direct commit does not. |
||
| } | ||
|
|
||
| RestAdapter adapter = new RestAdapter("dir", backendConfig, host, managedVersioning ? 0 : port); | ||
| adapter.start(); | ||
|
|
||
| ManagedVersioningProxy proxy = | ||
| managedVersioning ? new ManagedVersioningProxy(host, port, adapter.getPort()) : null; | ||
| if (proxy != null) { | ||
| proxy.start(); | ||
| } | ||
|
|
||
| Runtime.getRuntime() | ||
| .addShutdownHook( | ||
| new Thread( | ||
| () -> { | ||
| if (proxy != null) { | ||
| proxy.close(); | ||
| } | ||
| adapter.close(); | ||
| })); | ||
| System.out.printf( | ||
| "Lance REST directory namespace listening on http://%s:%d with root %s " | ||
| + "(managed versioning: %s)%n", | ||
| host, managedVersioning ? port : adapter.getPort(), root, managedVersioning); | ||
| new CountDownLatch(1).await(); | ||
| } | ||
|
|
||
| /** | ||
| * Proxies the managed REST test server so tests can observe successful namespace-owned version | ||
| * commits. Direct Dataset commits bypass this proxy endpoint and therefore do not increment the | ||
| * counter. | ||
| */ | ||
| private static final class ManagedVersioningProxy implements AutoCloseable { | ||
| private static final Set<String> HOP_BY_HOP_HEADERS = | ||
| Set.of("connection", "content-length", "expect", "host", "transfer-encoding", "upgrade"); | ||
|
|
||
| private final String backendUri; | ||
| private final HttpClient client = HttpClient.newHttpClient(); | ||
| private final AtomicLong createTableVersionCount = new AtomicLong(); | ||
| private final ExecutorService executor = Executors.newCachedThreadPool(); | ||
| private final HttpServer server; | ||
|
|
||
| private ManagedVersioningProxy(String host, int port, int backendPort) throws IOException { | ||
| this.backendUri = "http://" + host + ":" + backendPort; | ||
| this.server = HttpServer.create(new InetSocketAddress(host, port), 0); | ||
| this.server.createContext("/", this::handle); | ||
| this.server.setExecutor(executor); | ||
| } | ||
|
|
||
| private void start() { | ||
| server.start(); | ||
| } | ||
|
|
||
| private void handle(HttpExchange exchange) throws IOException { | ||
| try { | ||
| if (CREATE_TABLE_VERSION_COUNT_PATH.equals(exchange.getRequestURI().getPath())) { | ||
| handleCreateTableVersionCount(exchange); | ||
| return; | ||
| } | ||
|
|
||
| byte[] requestBody = exchange.getRequestBody().readAllBytes(); | ||
| URI target = URI.create(backendUri + exchange.getRequestURI().toASCIIString()); | ||
| HttpRequest.Builder request = | ||
| HttpRequest.newBuilder(target) | ||
| .method( | ||
| exchange.getRequestMethod(), | ||
| HttpRequest.BodyPublishers.ofByteArray(requestBody)); | ||
| exchange | ||
| .getRequestHeaders() | ||
| .forEach( | ||
| (name, values) -> { | ||
| if (!isHopByHopHeader(name)) { | ||
| values.forEach(value -> request.header(name, value)); | ||
| } | ||
| }); | ||
|
|
||
| HttpResponse<byte[]> response; | ||
| try { | ||
| response = client.send(request.build(), HttpResponse.BodyHandlers.ofByteArray()); | ||
| } catch (IOException e) { | ||
| sendResponse( | ||
| exchange, | ||
| 502, | ||
| ("REST proxy failed: " + errorMessage(e)).getBytes(StandardCharsets.UTF_8)); | ||
| return; | ||
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| sendResponse(exchange, 502, "REST proxy interrupted".getBytes(StandardCharsets.UTF_8)); | ||
| return; | ||
| } | ||
|
|
||
| response | ||
| .headers() | ||
| .map() | ||
| .forEach( | ||
| (name, values) -> { | ||
| if (!isHopByHopHeader(name)) { | ||
| values.forEach(value -> exchange.getResponseHeaders().add(name, value)); | ||
| } | ||
| }); | ||
| if (isSuccessfulCreateTableVersion(exchange, response.statusCode())) { | ||
| createTableVersionCount.incrementAndGet(); | ||
| } | ||
| sendResponse(exchange, response.statusCode(), response.body()); | ||
| } catch (RuntimeException e) { | ||
| sendResponse( | ||
| exchange, | ||
| 502, | ||
| ("REST proxy failed: " + errorMessage(e)).getBytes(StandardCharsets.UTF_8)); | ||
| } finally { | ||
| exchange.close(); | ||
| } | ||
| } | ||
|
|
||
| private void handleCreateTableVersionCount(HttpExchange exchange) throws IOException { | ||
| if (!"GET".equals(exchange.getRequestMethod())) { | ||
| sendResponse(exchange, 405, new byte[0]); | ||
| return; | ||
| } | ||
| exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8"); | ||
| sendResponse( | ||
| exchange, | ||
| 200, | ||
| Long.toString(createTableVersionCount.get()).getBytes(StandardCharsets.UTF_8)); | ||
| } | ||
|
|
||
| private static boolean isSuccessfulCreateTableVersion(HttpExchange exchange, int statusCode) { | ||
| String path = exchange.getRequestURI().getPath(); | ||
| return "POST".equals(exchange.getRequestMethod()) | ||
| && path.startsWith("/v1/table/") | ||
| && path.endsWith("/version/create") | ||
| && statusCode >= 200 | ||
| && statusCode < 300; | ||
| } | ||
|
|
||
| private static boolean isHopByHopHeader(String name) { | ||
| return HOP_BY_HOP_HEADERS.contains(name.toLowerCase()); | ||
| } | ||
|
|
||
| private static String errorMessage(Exception error) { | ||
| return error.getMessage() == null ? error.getClass().getSimpleName() : error.getMessage(); | ||
| } | ||
|
|
||
| private static void sendResponse(HttpExchange exchange, int statusCode, byte[] body) | ||
| throws IOException { | ||
| exchange.sendResponseHeaders(statusCode, body.length); | ||
| exchange.getResponseBody().write(body); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| server.stop(0); | ||
| executor.shutdownNow(); | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This new managed test step is not triggered by changes to the backfill implementation it protects. The workflow's
pull_request.pathsremains search-focused;LanceDataset.java,AddColumnsBackfillBatchWrite.java, andUpdateColumnsBackfillBatchWrite.javamatch none of its entries. The general Spark workflow does run for those paths, but its default backends skip theserequires_resttests, so a production-only regression receives no managed backfill run.Please add the relevant catalog/backfill production paths to this workflow's trigger, or move the managed test into a workflow whose PR trigger already covers them.
Reproducer
All three results were
False.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The Java paths are now covered, but the managed test still does not run for AddColumnsBackfillExec.scala or UpdateColumnsBackfillExec.scala. Those executors propagate managedVersioning into the temporary LanceDataset, so changing that propagation can reintroduce the failure without triggering this workflow.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in c63ca60: the workflow path filter now matches both AddColumnsBackfillExec.scala and UpdateColumnsBackfillExec.scala, so changes to either managed-versioning propagation path select the managed REST backfill test.