From 02f7dbebcf533a5da5133471053c72848824b59c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9lder=20Greg=C3=B3rio?= Date: Thu, 10 Sep 2026 21:06:29 +0100 Subject: [PATCH 1/5] BDX-645: prototype PollInfo support for JDBC --- .../driver/jdbc/ArrowFlightConnection.java | 1 + .../ArrowFlightJdbcFlightStreamResultSet.java | 54 ++- .../driver/jdbc/ArrowFlightMetaImpl.java | 11 +- .../jdbc/ArrowFlightPreparedStatement.java | 30 +- .../driver/jdbc/ArrowFlightStatement.java | 30 +- .../client/ArrowFlightSqlClientHandler.java | 179 +++++++- .../driver/jdbc/client/PollInfoOperation.java | 129 ++++++ .../ArrowFlightConnectionConfigImpl.java | 6 + .../driver/jdbc/OAuthIntegrationTest.java | 2 +- .../driver/jdbc/PollInfoExecutionTest.java | 418 ++++++++++++++++++ .../SharedServerPollInfoExecutionTest.java | 353 +++++++++++++++ .../jdbc/utils/MockFlightSqlProducer.java | 2 +- .../utils/PollingMockFlightSqlProducer.java | 256 +++++++++++ .../arrow/flight/sql/FlightSqlClient.java | 46 +- reports/call-site-inventory.md | 27 ++ reports/evidence.jsonl | 10 + reports/jdbc-report.md | 169 +++++++ 17 files changed, 1693 insertions(+), 30 deletions(-) create mode 100644 flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java create mode 100644 flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java create mode 100644 flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java create mode 100644 flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java create mode 100644 reports/call-site-inventory.md create mode 100644 reports/evidence.jsonl create mode 100644 reports/jdbc-report.md diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java index 623c2b81be..5fe70009c7 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java @@ -126,6 +126,7 @@ private static ArrowFlightSqlClientHandler createNewClientHandler( .withCatalog(config.getCatalog()) .withClientCache(config.useClientCache() ? new FlightClientCache() : null) .withConnectTimeout(config.getConnectTimeout()) + .withPollInfo(config.usePollInfo()) .withDriverVersion(driverVersion) .withOAuthConfiguration(config.getOauthConfiguration()) .build(); diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java index 376e5b11e7..91f9a5a8ca 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java @@ -21,6 +21,7 @@ import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.SQLException; +import java.sql.SQLTimeoutException; import java.util.Optional; import java.util.TimeZone; import java.util.concurrent.TimeUnit; @@ -28,6 +29,8 @@ import org.apache.arrow.driver.jdbc.utils.FlightEndpointDataQueue; import org.apache.arrow.driver.jdbc.utils.VectorSchemaRootTransformer; import org.apache.arrow.flight.FlightInfo; +import org.apache.arrow.flight.FlightRuntimeException; +import org.apache.arrow.flight.FlightStatusCode; import org.apache.arrow.flight.FlightStream; import org.apache.arrow.util.AutoCloseables; import org.apache.arrow.vector.VectorSchemaRoot; @@ -67,7 +70,20 @@ public final class ArrowFlightJdbcFlightStreamResultSet throws SQLException { super(statement, state, signature, resultSetMetaData, timeZone, firstFrame); this.connection = (ArrowFlightConnection) statement.connection; - this.flightInfo = ((ArrowFlightInfoStatement) statement).executeFlightInfoQuery(); + try { + this.flightInfo = ((ArrowFlightInfoStatement) statement).executeFlightInfoQuery(); + } catch (FlightRuntimeException e) { + if (e.status().code() != FlightStatusCode.TIMED_OUT) { + throw e; + } + final SQLTimeoutException jdbcTimeout = + new SQLTimeoutException( + String.format( + "Query timed out after %d %s", + statement.getQueryTimeout(), TimeUnit.SECONDS)); + jdbcTimeout.initCause(e); + throw jdbcTimeout; + } } /** Private constructor for fromFlightInfo. */ @@ -260,12 +276,40 @@ public synchronized void close() { private CloseableEndpointStreamPair getNextEndpointStream(final boolean canTimeout) throws SQLException { if (canTimeout) { - final int statementTimeout = statement != null ? statement.getQueryTimeout() : 0; - return statementTimeout != 0 - ? flightEndpointDataQueue.next(statementTimeout, TimeUnit.SECONDS) - : flightEndpointDataQueue.next(); + final long remainingTimeoutNanos = remainingQueryTimeoutNanos(); + if (remainingTimeoutNanos != Long.MAX_VALUE) { + if (remainingTimeoutNanos <= 0) { + throw new SQLTimeoutException("Query timed out before retrieving its first endpoint"); + } + try { + return flightEndpointDataQueue.next(remainingTimeoutNanos, TimeUnit.NANOSECONDS); + } catch (SQLTimeoutException e) { + if (statement != null && e.getMessage().startsWith("Query timed out after")) { + final SQLTimeoutException jdbcTimeout = + new SQLTimeoutException( + String.format( + "Query timed out after %d %s", + statement.getQueryTimeout(), TimeUnit.SECONDS)); + jdbcTimeout.initCause(e); + throw jdbcTimeout; + } + throw e; + } + } } else { return flightEndpointDataQueue.next(); } + return flightEndpointDataQueue.next(); + } + + private long remainingQueryTimeoutNanos() throws SQLException { + if (statement instanceof ArrowFlightStatement) { + return ((ArrowFlightStatement) statement).remainingQueryTimeoutNanos(); + } + if (statement instanceof ArrowFlightPreparedStatement) { + return ((ArrowFlightPreparedStatement) statement).remainingQueryTimeoutNanos(); + } + final int statementTimeout = statement != null ? statement.getQueryTimeout() : 0; + return statementTimeout > 0 ? TimeUnit.SECONDS.toNanos(statementTimeout) : Long.MAX_VALUE; } } diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java index 0d85b5eddb..7dee01b0d6 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java @@ -181,9 +181,12 @@ public Frame fetch( String.format("%s does not use frames.", this), AvaticaConnection.HELPER.unsupported()); } - private PreparedStatement prepareForHandle(final String query, StatementHandle handle) { + private PreparedStatement prepareForHandle( + final String query, StatementHandle handle, final boolean directExecution) { final PreparedStatement preparedStatement = - ((ArrowFlightConnection) connection).getClientHandler().prepare(query); + directExecution + ? ((ArrowFlightConnection) connection).getClientHandler().prepareDirect(query) + : ((ArrowFlightConnection) connection).getClientHandler().prepare(query); handle.signature = newSignature( query, @@ -198,7 +201,7 @@ private PreparedStatement prepareForHandle(final String query, StatementHandle h public StatementHandle prepare( final ConnectionHandle connectionHandle, final String query, final long maxRowCount) { final StatementHandle handle = super.createStatement(connectionHandle); - prepareForHandle(query, handle); + prepareForHandle(query, handle, false); return handle; } @@ -222,7 +225,7 @@ public ExecuteResult prepareAndExecute( final PrepareCallback callback) throws NoSuchStatementException { try { - PreparedStatement preparedStatement = prepareForHandle(query, handle); + PreparedStatement preparedStatement = prepareForHandle(query, handle, true); final StatementType statementType = preparedStatement.getType(); final long updateCount = diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java index d7af6902f4..38c592c7bb 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java @@ -18,7 +18,9 @@ import java.sql.PreparedStatement; import java.sql.SQLException; +import java.util.concurrent.atomic.AtomicReference; import org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler; +import org.apache.arrow.driver.jdbc.client.PollInfoOperation; import org.apache.arrow.flight.FlightInfo; import org.apache.arrow.util.Preconditions; import org.apache.calcite.avatica.AvaticaPreparedStatement; @@ -30,6 +32,8 @@ public class ArrowFlightPreparedStatement extends AvaticaPreparedStatement implements ArrowFlightInfoStatement { private final ArrowFlightSqlClientHandler.PreparedStatement preparedStatement; + private final AtomicReference activeOperation = new AtomicReference<>(); + private volatile PollInfoOperation lastOperation; private ArrowFlightPreparedStatement( final ArrowFlightConnection connection, @@ -76,6 +80,30 @@ public synchronized void close() throws SQLException { @Override public FlightInfo executeFlightInfoQuery() throws SQLException { - return preparedStatement.executeQuery(); + final PollInfoOperation operation = new PollInfoOperation(getQueryTimeout()); + activeOperation.set(operation); + lastOperation = operation; + try { + return preparedStatement.executeQuery(operation); + } finally { + activeOperation.compareAndSet(operation, null); + operation.close(); + } + } + + @Override + public void cancel() throws SQLException { + final PollInfoOperation operation = activeOperation.get(); + if (operation != null) { + operation.cancel(); + } + super.cancel(); + } + + long remainingQueryTimeoutNanos() { + final PollInfoOperation operation = lastOperation; + return operation != null && operation.hasDeadline() + ? operation.remainingTimeoutNanos() + : Long.MAX_VALUE; } } diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java index ff3d060c50..623ea3179f 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java @@ -17,7 +17,9 @@ package org.apache.arrow.driver.jdbc; import java.sql.SQLException; +import java.util.concurrent.atomic.AtomicReference; import org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler.PreparedStatement; +import org.apache.arrow.driver.jdbc.client.PollInfoOperation; import org.apache.arrow.driver.jdbc.utils.ConvertUtils; import org.apache.arrow.flight.FlightInfo; import org.apache.arrow.vector.types.pojo.Schema; @@ -27,6 +29,8 @@ /** A SQL statement for querying data from an Arrow Flight server. */ public class ArrowFlightStatement extends AvaticaStatement implements ArrowFlightInfoStatement { + private final AtomicReference activeOperation = new AtomicReference<>(); + private volatile PollInfoOperation lastOperation; ArrowFlightStatement( final ArrowFlightConnection connection, @@ -57,6 +61,30 @@ public FlightInfo executeFlightInfoQuery() throws SQLException { ConvertUtils.convertArrowFieldsToColumnMetaDataList(resultSetSchema.getFields())); setSignature(signature); - return preparedStatement.executeQuery(); + final PollInfoOperation operation = new PollInfoOperation(getQueryTimeout()); + activeOperation.set(operation); + lastOperation = operation; + try { + return preparedStatement.executeQuery(operation); + } finally { + activeOperation.compareAndSet(operation, null); + operation.close(); + } + } + + @Override + public void cancel() throws SQLException { + final PollInfoOperation operation = activeOperation.get(); + if (operation != null) { + operation.cancel(); + } + super.cancel(); + } + + long remainingQueryTimeoutNanos() { + final PollInfoOperation operation = lastOperation; + return operation != null && operation.hasDeadline() + ? operation.remainingTimeoutNanos() + : Long.MAX_VALUE; } } diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java index 719cc38a2b..47374ebde4 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java @@ -17,6 +17,8 @@ package org.apache.arrow.driver.jdbc.client; import com.google.common.collect.ImmutableMap; +import com.google.protobuf.Any; +import com.google.protobuf.InvalidProtocolBufferException; import io.grpc.netty.NettyChannelBuilder; import io.netty.channel.ChannelOption; import java.io.IOException; @@ -32,6 +34,9 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; import org.apache.arrow.driver.jdbc.client.oauth.OAuthConfiguration; import org.apache.arrow.driver.jdbc.client.oauth.OAuthCredentialWriter; import org.apache.arrow.driver.jdbc.client.oauth.OAuthTokenProvider; @@ -40,6 +45,7 @@ import org.apache.arrow.driver.jdbc.client.utils.FlightLocationQueue; import org.apache.arrow.flight.CallOption; import org.apache.arrow.flight.CallStatus; +import org.apache.arrow.flight.CancelFlightInfoRequest; import org.apache.arrow.flight.CloseSessionRequest; import org.apache.arrow.flight.FlightClient; import org.apache.arrow.flight.FlightClientMiddleware; @@ -50,6 +56,7 @@ import org.apache.arrow.flight.FlightStatusCode; import org.apache.arrow.flight.Location; import org.apache.arrow.flight.LocationSchemes; +import org.apache.arrow.flight.PollInfo; import org.apache.arrow.flight.SessionOptionValueFactory; import org.apache.arrow.flight.SetSessionOptionsRequest; import org.apache.arrow.flight.SetSessionOptionsResult; @@ -120,7 +127,12 @@ static ArrowFlightSqlClientHandler createNewHandler( final @Nullable FlightClientCache flightClientCache) { final ArrowFlightSqlClientHandler handler = new ArrowFlightSqlClientHandler( - cacheKey, new FlightSqlClient(client), builder, options, catalog, flightClientCache); + cacheKey, + new PollingFlightSqlClient(client, builder.usePollInfo), + builder, + options, + catalog, + flightClientCache); handler.setSetCatalogInSessionIfPresent(); return handler; } @@ -360,6 +372,9 @@ public interface PreparedStatement extends AutoCloseable { */ FlightInfo executeQuery() throws SQLException; + /** Execute using a statement-owned timeout/cancellation operation. */ + FlightInfo executeQuery(PollInfoOperation operation) throws SQLException; + /** * Executes a {@link StatementType#UPDATE} query. * @@ -449,12 +464,42 @@ public void setCatalog(final String catalog) throws SQLException { * @return a new prepared statement. */ public PreparedStatement prepare(final String query) { + return prepare(query, false); + } + + /** Prepare for schema discovery, but execute through CommandStatementQuery. */ + public PreparedStatement prepareDirect(final String query) { + // The opt-out must preserve the exact legacy JDBC wire path, which executes direct JDBC + // statements through CommandPreparedStatementQuery. + return prepare(query, builder.usePollInfo); + } + + private PreparedStatement prepare(final String query, final boolean directExecution) { final FlightSqlClient.PreparedStatement preparedStatement = sqlClient.prepare(query, getOptions()); return new PreparedStatement() { @Override public FlightInfo executeQuery() throws SQLException { - return preparedStatement.execute(getOptions()); + return directExecution + ? sqlClient.execute(query, getOptions()) + : preparedStatement.execute(getOptions()); + } + + @Override + public FlightInfo executeQuery(final PollInfoOperation operation) throws SQLException { + if (sqlClient instanceof PollingFlightSqlClient) { + final CallOption[] operationOptions = operation.options(getOptions()); + return ((PollingFlightSqlClient) sqlClient) + .withOperation( + operation, + () -> + directExecution + ? sqlClient.execute(query, operationOptions) + : preparedStatement.execute(operationOptions)); + } + return directExecution + ? sqlClient.execute(query, getOptions()) + : preparedStatement.execute(getOptions()); } @Override @@ -653,6 +698,127 @@ public FlightInfo getCrossReference( getOptions()); } + /** Flight SQL client that resolves descriptors through synchronous PollFlightInfo. */ + private static final class PollingFlightSqlClient extends FlightSqlClient { + private static final String UNKNOWN_COMMAND_FAMILY = "unknown-command"; + + private final FlightClient client; + private final boolean pollInfoEnabled; + private final Set unsupportedFamilies = ConcurrentHashMap.newKeySet(); + private final ThreadLocal currentOperation = new ThreadLocal<>(); + + private PollingFlightSqlClient(final FlightClient client, final boolean pollInfoEnabled) { + super(client); + this.client = client; + this.pollInfoEnabled = pollInfoEnabled; + } + + private FlightInfo withOperation( + final PollInfoOperation operation, final Supplier action) { + final PollInfoOperation previous = currentOperation.get(); + currentOperation.set(operation); + try { + // Enter the statement-owned cancellable context before prepared parameter upload as well + // as descriptor polling. This lets timeout/cancel terminate a blocked DoPut. + return operation.call(action::get); + } finally { + if (previous == null) { + currentOperation.remove(); + } else { + currentOperation.set(previous); + } + } + } + + @Override + protected FlightInfo getInfo( + final org.apache.arrow.flight.FlightDescriptor originalDescriptor, + final CallOption... options) { + PollInfoOperation operation = currentOperation.get(); + final boolean ownsOperation = operation == null; + if (ownsOperation) { + operation = new PollInfoOperation(0); + } + try { + return getInfo(operation, originalDescriptor, options); + } finally { + if (ownsOperation) { + operation.close(); + } + } + } + + private FlightInfo getInfo( + final PollInfoOperation operation, + final org.apache.arrow.flight.FlightDescriptor originalDescriptor, + final CallOption[] options) { + final String family = commandFamily(originalDescriptor); + if (!pollInfoEnabled || unsupportedFamilies.contains(family)) { + return operation.call(() -> client.getInfo(originalDescriptor, operation.options(options))); + } + + org.apache.arrow.flight.FlightDescriptor descriptor = originalDescriptor; + boolean initialPoll = true; + try { + while (true) { + final org.apache.arrow.flight.FlightDescriptor descriptorForPoll = descriptor; + final PollInfo pollInfo = + operation.call(() -> client.pollInfo(descriptorForPoll, operation.options(options))); + operation.remember(pollInfo.getFlightInfo()); + if (!pollInfo.getFlightDescriptor().isPresent()) { + return pollInfo.getFlightInfo(); + } + descriptor = pollInfo.getFlightDescriptor().get(); + initialPoll = false; + } + } catch (FlightRuntimeException e) { + if (initialPoll && e.status().code() == FlightStatusCode.UNIMPLEMENTED) { + unsupportedFamilies.add(family); + return operation.call( + () -> client.getInfo(originalDescriptor, operation.options(options))); + } + if (operation.isCancelled() + || e.status().code() == FlightStatusCode.CANCELLED + || e.status().code() == FlightStatusCode.TIMED_OUT) { + cancelFlightInfoBestEffort(operation, options); + } + throw e; + } + } + + private void cancelFlightInfoBestEffort( + final PollInfoOperation operation, final CallOption[] options) { + if (!operation.beginCancelFlightInfoAttempt()) { + return; + } + final CallOption[] cancelOptions = Arrays.copyOf(options, options.length + 1); + cancelOptions[options.length] = + org.apache.arrow.flight.CallOptions.timeout(1, TimeUnit.SECONDS); + try { + // The primary statement context is already cancelled/timed out. Use a fresh root context + // for this bounded best-effort cleanup action so it can reach the server. + io.grpc.Context.ROOT.run( + () -> + client.cancelFlightInfo( + new CancelFlightInfoRequest(operation.latestFlightInfo()), cancelOptions)); + } catch (RuntimeException e) { + LOGGER.debug("Best-effort CancelFlightInfo failed", e); + } + } + + private static String commandFamily(final org.apache.arrow.flight.FlightDescriptor descriptor) { + if (!descriptor.isCommand()) { + return "path"; + } + try { + final String typeUrl = Any.parseFrom(descriptor.getCommand()).getTypeUrl(); + return typeUrl.isEmpty() ? UNKNOWN_COMMAND_FAMILY : typeUrl; + } catch (InvalidProtocolBufferException e) { + return UNKNOWN_COMMAND_FAMILY; + } + } + } + /** Builder for {@link ArrowFlightSqlClientHandler}. */ public static final class Builder { static final String USER_AGENT_TEMPLATE = "JDBC Flight SQL Driver %s"; @@ -697,6 +863,8 @@ public static final class Builder { @VisibleForTesting @Nullable Duration connectTimeout; + @VisibleForTesting boolean usePollInfo = true; + @VisibleForTesting @Nullable OAuthConfiguration oauthConfig; // These two middleware are for internal use within build() and should not be @@ -739,6 +907,7 @@ public Builder() {} this.allocator = original.allocator; this.catalog = original.catalog; this.oauthConfig = original.oauthConfig; + this.usePollInfo = original.usePollInfo; if (original.retainCookies) { this.cookieFactory = original.cookieFactory; @@ -997,6 +1166,12 @@ public Builder withConnectTimeout(Duration connectTimeout) { return this; } + /** Select PollFlightInfo (default) or legacy GetFlightInfo descriptor resolution. */ + public Builder withPollInfo(final boolean usePollInfo) { + this.usePollInfo = usePollInfo; + return this; + } + /** * Sets the driver version for this handler. * diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java new file mode 100644 index 0000000000..979bea1092 --- /dev/null +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java @@ -0,0 +1,129 @@ +/* + * 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.arrow.driver.jdbc.client; + +import io.grpc.Context; +import java.util.Arrays; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallOptions; +import org.apache.arrow.flight.CallStatus; +import org.apache.arrow.flight.FlightInfo; + +/** + * State shared by every PollFlightInfo call in one synchronous JDBC query execution. + * + *

The absolute deadline is deliberately created once. Each RPC receives only the remaining + * budget, while the cancellable gRPC context lets {@code Statement.cancel()} terminate a poll that + * is active before a ResultSet exists. + */ +public final class PollInfoOperation implements AutoCloseable { + private static final long NO_DEADLINE = Long.MAX_VALUE; + + private final long deadlineNanos; + private final Context.CancellableContext context; + private final AtomicBoolean cancelled = new AtomicBoolean(); + private final AtomicBoolean cancelFlightInfoAttempted = new AtomicBoolean(); + private volatile FlightInfo latestFlightInfo; + + /** Create an operation whose timeout is expressed in JDBC query-timeout seconds. */ + public PollInfoOperation(final int timeoutSeconds) { + deadlineNanos = + timeoutSeconds > 0 + ? saturatingAdd(System.nanoTime(), TimeUnit.SECONDS.toNanos(timeoutSeconds)) + : NO_DEADLINE; + context = Context.current().withCancellation(); + } + + /** Run one RPC in the operation's cancellable context. */ + T call(final Callable callable) { + if (cancelled.get()) { + throw CallStatus.CANCELLED.withDescription("Statement canceled").toRuntimeException(); + } + try { + return context.call(callable); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + /** Return the base options plus this operation's remaining deadline. */ + CallOption[] options(final CallOption[] baseOptions) { + if (deadlineNanos == NO_DEADLINE) { + return baseOptions; + } + final long remaining = remainingTimeoutNanos(); + if (remaining <= 0) { + throw CallStatus.TIMED_OUT.withDescription("JDBC query timeout expired").toRuntimeException(); + } + final CallOption[] options = Arrays.copyOf(baseOptions, baseOptions.length + 1); + options[baseOptions.length] = CallOptions.timeout(remaining, TimeUnit.NANOSECONDS); + return options; + } + + /** Cancel the active PollFlightInfo call, if any. */ + public void cancel() { + cancelled.set(true); + context.cancel(null); + } + + boolean isCancelled() { + return cancelled.get(); + } + + void remember(final FlightInfo flightInfo) { + latestFlightInfo = flightInfo; + } + + FlightInfo latestFlightInfo() { + return latestFlightInfo; + } + + boolean beginCancelFlightInfoAttempt() { + return latestFlightInfo != null && cancelFlightInfoAttempted.compareAndSet(false, true); + } + + /** Return the remaining operation-wide timeout, in nanoseconds. */ + public long remainingTimeoutNanos() { + if (deadlineNanos == NO_DEADLINE) { + return NO_DEADLINE; + } + return deadlineNanos - System.nanoTime(); + } + + /** Whether this operation has a finite deadline. */ + public boolean hasDeadline() { + return deadlineNanos != NO_DEADLINE; + } + + @Override + public void close() { + context.close(); + } + + private static long saturatingAdd(final long left, final long right) { + final long result = left + right; + if (((left ^ result) & (right ^ result)) < 0) { + return Long.MAX_VALUE; + } + return result; + } +} diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ArrowFlightConnectionConfigImpl.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ArrowFlightConnectionConfigImpl.java index d0ba74dbcc..3b3bc26dcf 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ArrowFlightConnectionConfigImpl.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ArrowFlightConnectionConfigImpl.java @@ -182,6 +182,11 @@ public boolean useClientCache() { return ArrowFlightConnectionProperty.USE_CLIENT_CACHE.getBoolean(properties); } + /** Whether normal FlightInfo-producing operations use PollFlightInfo. */ + public boolean usePollInfo() { + return ArrowFlightConnectionProperty.USE_POLL_INFO.getBoolean(properties); + } + /** * Gets the {@link CallOption}s from this {@link ConnectionConfig}. * @@ -267,6 +272,7 @@ public enum ArrowFlightConnectionProperty implements ConnectionProperty { CATALOG("catalog", null, Type.STRING, false), CONNECT_TIMEOUT_MILLIS("connectTimeoutMs", 10000, Type.NUMBER, false), USE_CLIENT_CACHE("useClientCache", true, Type.BOOLEAN, false), + USE_POLL_INFO("usePollInfo", true, Type.BOOLEAN, false), // OAuth configuration properties OAUTH_FLOW("oauth.flow", null, Type.STRING, false), diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/OAuthIntegrationTest.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/OAuthIntegrationTest.java index 5e782db031..e0d5d963f7 100644 --- a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/OAuthIntegrationTest.java +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/OAuthIntegrationTest.java @@ -467,7 +467,7 @@ public void testOAuthTokenSentAsBearer() throws Exception { String authHeader = FLIGHT_SERVER_TEST_EXTENSION .getInterceptorFactory() - .getHeader(org.apache.arrow.flight.FlightMethod.GET_FLIGHT_INFO, "authorization"); + .getHeader(org.apache.arrow.flight.FlightMethod.POLL_FLIGHT_INFO, "authorization"); assertNotNull(authHeader, "Authorization header should be present in Flight requests"); assertEquals("Bearer " + VALID_ACCESS_TOKEN, authHeader); } diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java new file mode 100644 index 0000000000..3a95af0ee6 --- /dev/null +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java @@ -0,0 +1,418 @@ +/* + * 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.arrow.driver.jdbc; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.protobuf.Any; +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLTimeoutException; +import java.sql.Statement; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.arrow.driver.jdbc.utils.PollingMockFlightSqlProducer; +import org.apache.arrow.driver.jdbc.utils.PollingMockFlightSqlProducer.Scenario; +import org.apache.arrow.flight.FlightDescriptor; +import org.apache.arrow.flight.FlightRuntimeException; +import org.apache.arrow.flight.FlightServer; +import org.apache.arrow.flight.FlightStatusCode; +import org.apache.arrow.flight.Location; +import org.apache.arrow.flight.sql.FlightSqlProducer; +import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs; +import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery; +import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery; +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.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.arrow.vector.util.Text; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** Executable POC coverage for synchronous PollInfo through the unchanged JDBC surface. */ +public class PollInfoExecutionTest { + private static final String DIRECT_QUERY = "SELECT 42 AS answer"; + private static final String PREPARED_QUERY = "SELECT ? AS answer"; + private static final Schema RESULT_SCHEMA = + new Schema(Collections.singletonList(Field.nullable("answer", new ArrowType.Int(32, true)))); + private static final String STATEMENT_FAMILY = + Any.pack(CommandStatementQuery.getDefaultInstance()).getTypeUrl(); + private static final String PREPARED_FAMILY = + Any.pack(CommandPreparedStatementQuery.getDefaultInstance()).getTypeUrl(); + private static final String CATALOGS_FAMILY = + Any.pack(CommandGetCatalogs.getDefaultInstance()).getTypeUrl(); + + private static BufferAllocator allocator; + private static PollingMockFlightSqlProducer producer; + private static FlightServer server; + + @BeforeAll + public static void startServer() throws Exception { + allocator = new RootAllocator(Long.MAX_VALUE); + producer = new PollingMockFlightSqlProducer(); + producer.addSelectQuery( + DIRECT_QUERY, RESULT_SCHEMA, Collections.singletonList(PollInfoExecutionTest::send42)); + producer.addSelectQuery( + PREPARED_QUERY, RESULT_SCHEMA, Collections.singletonList(PollInfoExecutionTest::send7)); + producer.addExpectedParameters( + PREPARED_QUERY, + new Schema( + Collections.singletonList(Field.nullable("parameter", new ArrowType.Int(32, true)))), + Collections.singletonList(Collections.singletonList(7))); + producer.addCatalogQuery( + CommandGetCatalogs.getDefaultInstance(), PollInfoExecutionTest::sendCatalog); + server = + FlightServer.builder(allocator, Location.forGrpcInsecure("localhost", 0), producer) + .build() + .start(); + } + + @AfterAll + public static void stopServer() throws Exception { + server.close(); + producer.close(); + allocator.close(); + } + + @Test + public void immediateCompletionUsesOnePollAndNoGetFlightInfo() throws Exception { + producer.configure(Scenario.IMMEDIATE, STATEMENT_FAMILY); + try (Connection connection = connect(true); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { + assertTrue(resultSet.next()); + assertEquals(42, resultSet.getInt(1)); + assertFalse(resultSet.next()); + } + assertEquals(1, producer.pollDescriptors().size()); + assertEquals(STATEMENT_FAMILY, family(producer.pollDescriptors().get(0))); + assertEquals(0, producer.getFlightInfoCount()); + } + + @Test + public void multiStepUsesEachDescriptorOnceAndOnlyFinalInfoFeedsResultSet() throws Exception { + producer.configure(Scenario.MULTI_STEP, STATEMENT_FAMILY); + try (Connection connection = connect(true); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { + assertTrue(resultSet.next()); + assertEquals(42, resultSet.getInt(1)); + } + final List descriptors = producer.pollDescriptors(); + assertEquals(3, descriptors.size()); + assertEquals(STATEMENT_FAMILY, family(descriptors.get(0))); + assertEquals( + "jdbc-continuation-1", new String(descriptors.get(1).getCommand(), StandardCharsets.UTF_8)); + assertEquals( + "jdbc-continuation-2", new String(descriptors.get(2).getCommand(), StandardCharsets.UTF_8)); + assertEquals(3, descriptors.stream().distinct().count()); + assertEquals(0, producer.getFlightInfoCount()); + } + + @Test + public void preparedParametersAreBoundOnceBeforePolling() throws Exception { + producer.configure(Scenario.MULTI_STEP, PREPARED_FAMILY); + try (Connection connection = connect(true); + PreparedStatement statement = connection.prepareStatement(PREPARED_QUERY)) { + statement.setInt(1, 7); + try (ResultSet resultSet = statement.executeQuery()) { + assertTrue(resultSet.next()); + assertEquals(7, resultSet.getInt(1)); + } + } + assertEquals(1, producer.parameterBindCount()); + assertEquals(3, producer.pollDescriptors().size()); + assertEquals(PREPARED_FAMILY, family(producer.pollDescriptors().get(0))); + assertEquals(0, producer.getFlightInfoCount()); + } + + @Test + public void metadataUsesPollInfoTransparently() throws Exception { + producer.configure(Scenario.MULTI_STEP, CATALOGS_FAMILY); + try (Connection connection = connect(true); + ResultSet resultSet = connection.getMetaData().getCatalogs()) { + assertTrue(resultSet.next()); + assertEquals("poll_catalog", resultSet.getString("TABLE_CAT")); + } + assertEquals(3, producer.pollDescriptors().size()); + assertEquals(CATALOGS_FAMILY, family(producer.pollDescriptors().get(0))); + assertEquals(0, producer.getFlightInfoCount()); + } + + @Test + public void unimplementedFallbackIsCachedByFamilyOnly() throws Exception { + producer.configure(Scenario.UNIMPLEMENTED, STATEMENT_FAMILY); + try (Connection connection = connect(true)) { + assertDirectRow(connection); + assertDirectRow(connection); + try (ResultSet resultSet = connection.getMetaData().getCatalogs()) { + assertTrue(resultSet.next()); + } + } + assertEquals(2, producer.pollDescriptors().size()); + assertEquals(STATEMENT_FAMILY, family(producer.pollDescriptors().get(0))); + assertEquals(CATALOGS_FAMILY, family(producer.pollDescriptors().get(1))); + assertEquals(2, producer.getFlightInfoCount()); + } + + @Test + public void connectionOptOutUsesOnlyGetFlightInfo() throws Exception { + producer.configure(Scenario.IMMEDIATE, PREPARED_FAMILY); + try (Connection connection = connect(false)) { + assertDirectRow(connection); + } + assertEquals(0, producer.pollDescriptors().size()); + assertEquals(1, producer.getFlightInfoCount()); + assertEquals(PREPARED_FAMILY, producer.lastGetFlightInfoFamily()); + } + + @Test + public void unavailableDoesNotFallbackOrReexecute() throws Exception { + producer.configure(Scenario.UNAVAILABLE, STATEMENT_FAMILY); + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + final Throwable failure = + assertThrows(Throwable.class, () -> statement.executeQuery(DIRECT_QUERY)); + assertTrue(hasFlightStatus(failure, FlightStatusCode.UNAVAILABLE), failure.toString()); + } + assertEquals(1, producer.pollDescriptors().size()); + assertEquals(0, producer.getFlightInfoCount()); + } + + @Test + public void continuationUnimplementedDoesNotFallbackOrReexecute() throws Exception { + producer.configure(Scenario.CONTINUATION_UNIMPLEMENTED, STATEMENT_FAMILY); + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + final Throwable failure = + assertThrows(Throwable.class, () -> statement.executeQuery(DIRECT_QUERY)); + assertTrue(hasFlightStatus(failure, FlightStatusCode.UNIMPLEMENTED), failure.toString()); + } + assertEquals(2, producer.pollDescriptors().size()); + assertEquals(0, producer.getFlightInfoCount()); + } + + @Test + public void oneDeadlineSpansAllPollsAndTerminatesActivePoll() throws Exception { + producer.configure(Scenario.TIMEOUT, STATEMENT_FAMILY); + final long startNanos = System.nanoTime(); + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + statement.setQueryTimeout(1); + final Throwable failure = + assertThrows(Throwable.class, () -> statement.executeQuery(DIRECT_QUERY)); + assertTrue(hasFlightStatus(failure, FlightStatusCode.TIMED_OUT), failure.toString()); + assertJdbcTimeoutContract(failure, 1); + } + final long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + System.out.println("LOCAL_T8_ELAPSED_MILLIS " + elapsedMillis); + assertTrue(elapsedMillis >= 850, "deadline fired too early: " + elapsedMillis); + assertTrue(elapsedMillis < 1500, "deadline appears to have reset: " + elapsedMillis); + assertEquals(2, producer.pollDescriptors().size()); + assertEquals(0, producer.getFlightInfoCount()); + assertTrue(producer.awaitPollTerminated(1, TimeUnit.SECONDS)); + } + + @Test + public void preparedBindSharesTheOperationDeadline() throws Exception { + producer.configure(Scenario.BIND_TIMEOUT, PREPARED_FAMILY); + final long startNanos = System.nanoTime(); + try (Connection connection = connect(true); + PreparedStatement statement = connection.prepareStatement(PREPARED_QUERY)) { + statement.setInt(1, 7); + statement.setQueryTimeout(1); + final Throwable failure = assertThrows(Throwable.class, statement::executeQuery); + assertTrue(hasFlightStatus(failure, FlightStatusCode.TIMED_OUT), failure.toString()); + assertJdbcTimeoutContract(failure, 1); + } + final long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + assertTrue(elapsedMillis >= 850, "bind deadline fired too early: " + elapsedMillis); + assertTrue(elapsedMillis < 1500, "bind deadline was not operation-wide: " + elapsedMillis); + assertEquals(1, producer.parameterBindCount()); + assertEquals(0, producer.pollDescriptors().size()); + assertTrue(producer.awaitPollTerminated(1, TimeUnit.SECONDS)); + } + + @Test + public void preparedBindIsCancelledBeforePolling() throws Exception { + producer.configure(Scenario.BIND_CANCEL, PREPARED_FAMILY); + try (Connection connection = connect(true); + PreparedStatement statement = connection.prepareStatement(PREPARED_QUERY)) { + statement.setInt(1, 7); + final AtomicReference failure = new AtomicReference<>(); + final Thread execution = + new Thread( + () -> { + try (ResultSet ignored = statement.executeQuery()) { + // A blocked bind must not produce a ResultSet. + } catch (Throwable e) { + failure.set(e); + } + }, + "jdbc-prepared-bind-cancellation-test"); + execution.start(); + assertTrue(producer.awaitActivePoll(2, TimeUnit.SECONDS)); + assertNull(statement.getResultSet()); + statement.cancel(); + execution.join(TimeUnit.SECONDS.toMillis(2)); + assertFalse(execution.isAlive()); + assertNotNull(failure.get()); + assertTrue( + hasFlightStatus(failure.get(), FlightStatusCode.CANCELLED), failure.get().toString()); + assertEquals(1, producer.parameterBindCount()); + assertEquals(0, producer.pollDescriptors().size()); + assertEquals(0, producer.cancelFlightInfoCount()); + assertTrue(producer.awaitPollTerminated(1, TimeUnit.SECONDS)); + } + } + + @Test + public void avaticaHasNoResultSetTargetButStatementContextCancelsActivePoll() throws Exception { + producer.configure(Scenario.CANCEL_OBSERVABLE, STATEMENT_FAMILY); + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + final AtomicReference failure = new AtomicReference<>(); + final AtomicBoolean resultSetCreated = new AtomicBoolean(); + final Thread execution = + new Thread( + () -> { + try (ResultSet ignored = statement.executeQuery(DIRECT_QUERY)) { + resultSetCreated.set(true); + } catch (Throwable e) { + failure.set(e); + } + }, + "jdbc-pollinfo-cancellation-test"); + execution.start(); + assertTrue(producer.awaitActivePoll(2, TimeUnit.SECONDS)); + // AvaticaStatement.cancel() can only reach openResultSet, which is still null at this point. + assertNull(statement.getResultSet()); + final long cancelStart = System.nanoTime(); + statement.cancel(); + execution.join(TimeUnit.SECONDS.toMillis(2)); + final long cancelMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - cancelStart); + System.out.println("LOCAL_T9_CANCEL_MILLIS " + cancelMillis); + assertFalse(execution.isAlive()); + assertFalse(resultSetCreated.get(), "ResultSet must not exist while polling is active"); + assertNotNull(failure.get()); + assertTrue( + hasFlightStatus(failure.get(), FlightStatusCode.CANCELLED), failure.get().toString()); + assertTrue(cancelMillis < 1000, "cancellation was not prompt: " + cancelMillis); + assertTrue(producer.awaitPollTerminated(1, TimeUnit.SECONDS)); + assertEquals(1, producer.cancelFlightInfoCount()); + assertEquals(0, producer.getFlightInfoCount()); + } + } + + private static Connection connect(final boolean usePollInfo) throws Exception { + return DriverManager.getConnection( + String.format( + "jdbc:arrow-flight-sql://localhost:%d?useEncryption=false&usePollInfo=%s", + server.getPort(), usePollInfo)); + } + + private static void assertDirectRow(final Connection connection) throws Exception { + try (Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { + assertTrue(resultSet.next()); + assertEquals(42, resultSet.getInt(1)); + } + } + + private static String family(final FlightDescriptor descriptor) { + return PollingMockFlightSqlProducer.commandFamily(descriptor); + } + + private static boolean hasFlightStatus( + final Throwable throwable, final FlightStatusCode expected) { + Throwable current = throwable; + while (current != null) { + if (current instanceof FlightRuntimeException + && ((FlightRuntimeException) current).status().code() == expected) { + return true; + } + current = current.getCause(); + } + return false; + } + + private static void assertJdbcTimeoutContract( + final Throwable throwable, final int timeoutSeconds) { + Throwable current = throwable; + while (current != null && !(current instanceof SQLTimeoutException)) { + current = current.getCause(); + } + assertNotNull(current, "Expected SQLTimeoutException in cause chain: " + throwable); + assertEquals( + String.format("Query timed out after %d %s", timeoutSeconds, TimeUnit.SECONDS), + current.getMessage()); + } + + private static void send42(final FlightSqlProducer.ServerStreamListener listener) { + sendInt(listener, 42); + } + + private static void send7(final FlightSqlProducer.ServerStreamListener listener) { + sendInt(listener, 7); + } + + private static void sendInt( + final FlightSqlProducer.ServerStreamListener listener, final int value) { + try (BufferAllocator streamAllocator = new RootAllocator(); + VectorSchemaRoot root = VectorSchemaRoot.create(RESULT_SCHEMA, streamAllocator)) { + root.allocateNew(); + ((IntVector) root.getVector("answer")).setSafe(0, value); + root.setRowCount(1); + listener.start(root); + listener.putNext(); + } finally { + listener.completed(); + } + } + + private static void sendCatalog(final FlightSqlProducer.ServerStreamListener listener) { + try (BufferAllocator streamAllocator = new RootAllocator(); + VectorSchemaRoot root = + VectorSchemaRoot.create( + FlightSqlProducer.Schemas.GET_CATALOGS_SCHEMA, streamAllocator)) { + root.allocateNew(); + ((VarCharVector) root.getVector("catalog_name")).setSafe(0, new Text("poll_catalog")); + root.setRowCount(1); + listener.start(root); + listener.putNext(); + } finally { + listener.completed(); + } + } +} diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java new file mode 100644 index 0000000000..bcfa116dcd --- /dev/null +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java @@ -0,0 +1,353 @@ +/* + * 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.arrow.driver.jdbc; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLTimeoutException; +import java.sql.Statement; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import org.apache.arrow.flight.FlightRuntimeException; +import org.apache.arrow.flight.FlightStatusCode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.MethodOrderer.OrderAnnotation; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; +import org.junit.jupiter.api.TestMethodOrder; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; + +/** Black-box BDX-645 contract checks against the standalone shared conformance server. */ +@EnabledIfSystemProperty(named = "pollinfo.shared.enabled", matches = "true") +@TestMethodOrder(OrderAnnotation.class) +public class SharedServerPollInfoExecutionTest { + private static final HttpClient HTTP_CLIENT = HttpClient.newHttpClient(); + private static final int FLIGHT_PORT = Integer.getInteger("pollinfo.shared.flightPort", 32347); + private static final int CONTROL_PORT = Integer.getInteger("pollinfo.shared.controlPort", 32348); + + @BeforeEach + public void resetServer() throws Exception { + post("/reset", ""); + } + + @AfterEach + public void printServerState(final TestInfo testInfo) throws Exception { + System.out.println("SHARED_SERVER_STATE " + testInfo.getDisplayName()); + System.out.println(state()); + } + + @Test + @Order(1) + public void t1Immediate() throws Exception { + try (Connection connection = connect(true)) { + assertEquals(Arrays.asList(1L, 2L), directValues(connection, "immediate")); + } + assertCounters(1, 0, 0, 1, 0, 0, 0, 1); + } + + @Test + @Order(2) + public void t2MultiStepContinuationsAndFinalCumulativeInfo() throws Exception { + try (Connection connection = connect(true)) { + assertEquals(Arrays.asList(1L, 2L, 3L), directValues(connection, "multi-step")); + } + final String state = state(); + assertCounters(state, 3, 0, 0, 1, 2, 0, 0, 3); + assertTrue(state.contains("bdx-645-poll/v1/")); + assertTrue(state.indexOf("/1\"") < state.indexOf("/2\""), state); + } + + @Test + @Order(3) + public void t3PreparedParametersBoundExactlyOnce() throws Exception { + try (Connection connection = connect(true); + PreparedStatement statement = connection.prepareStatement("prepared-multi-step")) { + statement.setLong(1, 41); + try (ResultSet resultSet = statement.executeQuery()) { + assertEquals(Arrays.asList(41L, 42L), values(resultSet)); + } + } + assertCounters(3, 0, 1, 1, 2, 0, 0, 1); + } + + @Test + @Order(4) + public void t4MetadataMultiStep() throws Exception { + try (Connection connection = connect(true); + ResultSet resultSet = connection.getMetaData().getCatalogs()) { + assertTrue(resultSet.next()); + assertEquals("bdx_catalog", resultSet.getString("TABLE_CAT")); + assertFalse(resultSet.next()); + } + assertCounters(3, 0, 0, 1, 2, 0, 0, 1); + } + + @Test + @Order(5) + public void t5InitialUnimplementedFallbackCacheIsFamilyIsolated() throws Exception { + try (Connection connection = connect(true)) { + assertEquals(Arrays.asList(1L, 2L), directValues(connection, "unimplemented")); + assertEquals(Arrays.asList(1L, 2L), directValues(connection, "unimplemented")); + try (ResultSet resultSet = connection.getMetaData().getCatalogs()) { + assertTrue(resultSet.next()); + assertEquals("bdx_catalog", resultSet.getString("TABLE_CAT")); + } + } + final String state = state(); + assertCounters(state, 4, 2, 0, 2, 2, 0, 0, 3); + assertTrue(state.contains("\"direct\""), state); + assertTrue(state.contains("\"metadata\""), state); + } + + @Test + @Order(6) + public void t6ConnectionOptOut() throws Exception { + try (Connection connection = connect(false)) { + assertEquals(Arrays.asList(1L, 2L), directValues(connection, "immediate")); + } + final String state = state(); + assertCounters(state, 0, 1, 0, 0, 0, 0, 0, 1); + assertTrue( + state + .replaceAll("\\s+", "") + .contains("\"prepared\":{\"poll_flight_info\":0,\"get_flight_info\":1"), + state); + } + + @Test + @Order(7) + public void t7UnavailableDoesNotFallbackOrReexecute() throws Exception { + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + final Throwable failure = + assertThrows(Throwable.class, () -> statement.executeQuery("unavailable")); + assertTrue(hasFlightStatus(failure, FlightStatusCode.UNAVAILABLE), failure.toString()); + } + assertCounters(1, 0, 0, 1, 0, 0, 0, 0); + } + + @Test + @Order(8) + public void t8TimeoutEndsBlockedPollAtTheSingleStatementDeadline() throws Exception { + final long startNanos = System.nanoTime(); + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + statement.setQueryTimeout(1); + final Throwable failure = + assertThrows(Throwable.class, () -> statement.executeQuery("blocked-poll")); + assertTrue(hasFlightStatus(failure, FlightStatusCode.TIMED_OUT), failure.toString()); + assertJdbcTimeoutContract(failure, 1); + } + final long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + System.out.println("SHARED_T8_ELAPSED_MILLIS " + elapsedMillis); + assertTrue(elapsedMillis >= 800, "deadline fired too early: " + elapsedMillis); + assertTrue(elapsedMillis < 1800, "deadline fired too late: " + elapsedMillis); + assertCounters(1, 0, 0, 1, 0, 0, 1, 0); + } + + @Test + @Order(9) + public void t9CancelBeforeResultSetCancelsPollAndFlightInfo() throws Exception { + try (Connection connection = connect(true); + Statement statement = connection.createStatement()) { + final AtomicReference failure = new AtomicReference<>(); + final AtomicBoolean resultSetCreated = new AtomicBoolean(); + final Thread execution = + new Thread( + () -> { + try (ResultSet ignored = statement.executeQuery("cancel-observable")) { + resultSetCreated.set(true); + } catch (Throwable e) { + failure.set(e); + } + }, + "shared-server-jdbc-cancel"); + execution.start(); + awaitCounter("poll_flight_info", 2, Duration.ofSeconds(3)); + assertNull(statement.getResultSet()); + final long cancelStart = System.nanoTime(); + statement.cancel(); + execution.join(TimeUnit.SECONDS.toMillis(2)); + final long cancelMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - cancelStart); + System.out.println("SHARED_T9_CANCEL_MILLIS " + cancelMillis); + assertFalse(execution.isAlive()); + assertFalse(resultSetCreated.get()); + assertNotNull(failure.get()); + assertTrue( + hasFlightStatus(failure.get(), FlightStatusCode.CANCELLED), failure.get().toString()); + assertTrue(cancelMillis < 1000, "cancellation was not prompt: " + cancelMillis); + } + awaitCounter("cancellation", 1, Duration.ofSeconds(2)); + assertCounters(2, 0, 0, 1, 1, 1, 1, 0); + } + + private static Connection connect(final boolean usePollInfo) throws Exception { + return DriverManager.getConnection( + String.format( + "jdbc:arrow-flight-sql://127.0.0.1:%d/?useEncryption=false&usePollInfo=%s", + FLIGHT_PORT, usePollInfo)); + } + + private static List directValues(final Connection connection, final String query) + throws Exception { + try (Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(query)) { + final List values = new ArrayList<>(); + while (resultSet.next()) { + assertEquals(query, resultSet.getString("scenario")); + values.add(resultSet.getLong("value")); + } + return values; + } + } + + private static List values(final ResultSet resultSet) throws Exception { + final List values = new ArrayList<>(); + while (resultSet.next()) { + values.add(resultSet.getLong("value")); + } + return values; + } + + private static void assertCounters( + final int poll, + final int get, + final int bind, + final int original, + final int continuation, + final int cancellation, + final int activeTermination, + final int doGet) + throws Exception { + assertCounters( + state(), poll, get, bind, original, continuation, cancellation, activeTermination, doGet); + } + + private static void assertCounters( + final String state, + final int poll, + final int get, + final int bind, + final int original, + final int continuation, + final int cancellation, + final int activeTermination, + final int doGet) { + assertEquals(poll, counter(state, "poll_flight_info"), state); + assertEquals(get, counter(state, "get_flight_info"), state); + assertEquals(bind, counter(state, "parameter_binding"), state); + assertEquals(original, counter(state, "original_descriptors"), state); + assertEquals(continuation, counter(state, "continuation_descriptors"), state); + assertEquals(cancellation, counter(state, "cancellation"), state); + assertEquals(activeTermination, counter(state, "active_call_terminations"), state); + assertEquals(doGet, counter(state, "do_get"), state); + } + + private static void awaitCounter(final String name, final int expected, final Duration timeout) + throws Exception { + final long deadline = System.nanoTime() + timeout.toNanos(); + do { + if (counter(state(), name) >= expected) { + return; + } + Thread.sleep(20); + } while (System.nanoTime() < deadline); + assertEquals(expected, counter(state(), name), state()); + } + + private static int counter(final String state, final String name) { + final Matcher matcher = + Pattern.compile("\\\"" + Pattern.quote(name) + "\\\"\\s*:\\s*(\\d+)").matcher(state); + assertTrue(matcher.find(), state); + return Integer.parseInt(matcher.group(1)); + } + + private static String state() throws Exception { + return request("GET", "/state", ""); + } + + private static void post(final String path, final String body) throws Exception { + request("POST", path, body); + } + + private static String request(final String method, final String path, final String body) + throws Exception { + final HttpRequest.Builder builder = + HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + CONTROL_PORT + path)) + .timeout(Duration.ofSeconds(2)); + if ("POST".equals(method)) { + builder.POST(HttpRequest.BodyPublishers.ofString(body)); + } else { + builder.GET(); + } + final HttpResponse response = + HTTP_CLIENT.send(builder.build(), HttpResponse.BodyHandlers.ofString()); + if ("POST".equals(method)) { + assertEquals(204, response.statusCode(), response.body()); + } else { + assertEquals(200, response.statusCode(), response.body()); + } + return response.body(); + } + + private static boolean hasFlightStatus( + final Throwable throwable, final FlightStatusCode expected) { + Throwable current = throwable; + while (current != null) { + if (current instanceof FlightRuntimeException + && ((FlightRuntimeException) current).status().code() == expected) { + return true; + } + current = current.getCause(); + } + return false; + } + + private static void assertJdbcTimeoutContract( + final Throwable throwable, final int timeoutSeconds) { + Throwable current = throwable; + while (current != null && !(current instanceof SQLTimeoutException)) { + current = current.getCause(); + } + assertNotNull(current, "Expected SQLTimeoutException in cause chain: " + throwable); + assertEquals( + String.format("Query timed out after %d %s", timeoutSeconds, TimeUnit.SECONDS), + current.getMessage()); + } +} diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java index 6627d91ab6..f90849b0bd 100644 --- a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java @@ -90,7 +90,7 @@ import org.checkerframework.checker.nullness.qual.Nullable; /** An ad-hoc {@link FlightSqlProducer} for tests. */ -public final class MockFlightSqlProducer implements FlightSqlProducer { +public class MockFlightSqlProducer implements FlightSqlProducer { private final Map>> queryResults = new HashMap<>(); private final Map> selectResultProviders = new HashMap<>(); diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java new file mode 100644 index 0000000000..e31cccbdf0 --- /dev/null +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java @@ -0,0 +1,256 @@ +/* + * 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.arrow.driver.jdbc.utils; + +import com.google.protobuf.Any; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.arrow.flight.CallStatus; +import org.apache.arrow.flight.CancelFlightInfoRequest; +import org.apache.arrow.flight.CancelStatus; +import org.apache.arrow.flight.FlightDescriptor; +import org.apache.arrow.flight.FlightEndpoint; +import org.apache.arrow.flight.FlightInfo; +import org.apache.arrow.flight.FlightStream; +import org.apache.arrow.flight.PollInfo; +import org.apache.arrow.flight.PutResult; +import org.apache.arrow.flight.Ticket; +import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery; + +/** Controllable in-process PollInfo producer used by the JDBC POC conformance tests. */ +public final class PollingMockFlightSqlProducer extends MockFlightSqlProducer { + public enum Scenario { + IMMEDIATE, + MULTI_STEP, + UNIMPLEMENTED, + CONTINUATION_UNIMPLEMENTED, + UNAVAILABLE, + TIMEOUT, + BIND_TIMEOUT, + BIND_CANCEL, + CANCEL_OBSERVABLE + } + + private static final String PREPARED_FAMILY = + Any.pack(CommandPreparedStatementQuery.getDefaultInstance()).getTypeUrl(); + + private static final FlightDescriptor CONTINUATION_1 = + FlightDescriptor.command("jdbc-continuation-1".getBytes(StandardCharsets.UTF_8)); + private static final FlightDescriptor CONTINUATION_2 = + FlightDescriptor.command("jdbc-continuation-2".getBytes(StandardCharsets.UTF_8)); + + private final List pollDescriptors = + Collections.synchronizedList(new ArrayList<>()); + private final AtomicInteger getFlightInfoCount = new AtomicInteger(); + private final AtomicInteger parameterBindCount = new AtomicInteger(); + private final AtomicInteger cancelFlightInfoCount = new AtomicInteger(); + private volatile String lastGetFlightInfoFamily; + private volatile Scenario scenario = Scenario.IMMEDIATE; + private volatile String scenarioFamily; + private volatile boolean scenarioSelectedForExecution; + private volatile FlightInfo finalFlightInfo; + private volatile CountDownLatch activePoll = new CountDownLatch(1); + private volatile CountDownLatch pollTerminated = new CountDownLatch(1); + + public void configure(final Scenario scenario, final String scenarioFamily) { + this.scenario = scenario; + this.scenarioFamily = scenarioFamily; + pollDescriptors.clear(); + getFlightInfoCount.set(0); + parameterBindCount.set(0); + cancelFlightInfoCount.set(0); + lastGetFlightInfoFamily = null; + scenarioSelectedForExecution = false; + finalFlightInfo = null; + activePoll = new CountDownLatch(1); + pollTerminated = new CountDownLatch(1); + } + + @Override + public FlightInfo getFlightInfo(final CallContext context, final FlightDescriptor descriptor) { + getFlightInfoCount.incrementAndGet(); + lastGetFlightInfoFamily = commandFamily(descriptor); + return super.getFlightInfo(context, descriptor); + } + + @Override + public PollInfo pollFlightInfo(final CallContext context, final FlightDescriptor descriptor) { + pollDescriptors.add(descriptor); + final boolean initial = isFlightSqlCommand(descriptor); + if (initial) { + finalFlightInfo = super.getFlightInfo(context, descriptor); + scenarioSelectedForExecution = + scenarioFamily == null || scenarioFamily.equals(commandFamily(descriptor)); + } + + final boolean selectedFamily = scenarioSelectedForExecution; + if (initial && selectedFamily && scenario == Scenario.UNIMPLEMENTED) { + throw CallStatus.UNIMPLEMENTED.toRuntimeException(); + } + if (initial && selectedFamily && scenario == Scenario.UNAVAILABLE) { + throw CallStatus.UNAVAILABLE.toRuntimeException(); + } + if (selectedFamily && scenario == Scenario.CONTINUATION_UNIMPLEMENTED) { + if (initial) { + return new PollInfo(partialInfo(finalFlightInfo), CONTINUATION_1, 0.25, null); + } + throw CallStatus.UNIMPLEMENTED.toRuntimeException(); + } + if (selectedFamily && scenario == Scenario.TIMEOUT) { + if (initial) { + sleep(700); + return new PollInfo(partialInfo(finalFlightInfo), CONTINUATION_1, 0.25, null); + } + return blockUntilCancelled(context); + } + if (selectedFamily && scenario == Scenario.CANCEL_OBSERVABLE) { + if (initial) { + return new PollInfo(partialInfo(finalFlightInfo), CONTINUATION_1, 0.25, null); + } + return blockUntilCancelled(context); + } + if (selectedFamily && scenario == Scenario.MULTI_STEP) { + if (initial) { + return new PollInfo(partialInfo(finalFlightInfo), CONTINUATION_1, 0.25, null); + } + if (descriptor.equals(CONTINUATION_1)) { + return new PollInfo(partialInfo(finalFlightInfo), CONTINUATION_2, 0.75, null); + } + } + return new PollInfo(finalFlightInfo, null, 1.0, null); + } + + @Override + public Runnable acceptPutPreparedStatementQuery( + final CommandPreparedStatementQuery command, + final CallContext context, + final FlightStream stream, + final StreamListener listener) { + parameterBindCount.incrementAndGet(); + final Runnable delegate = + super.acceptPutPreparedStatementQuery(command, context, stream, listener); + if (PREPARED_FAMILY.equals(scenarioFamily) + && (scenario == Scenario.BIND_TIMEOUT || scenario == Scenario.BIND_CANCEL)) { + return () -> { + activePoll.countDown(); + try { + while (!context.isCancelled()) { + sleep(10); + } + throw CallStatus.CANCELLED.toRuntimeException(); + } finally { + pollTerminated.countDown(); + } + }; + } + return delegate; + } + + @Override + public void cancelFlightInfo( + final CancelFlightInfoRequest request, + final CallContext context, + final StreamListener listener) { + cancelFlightInfoCount.incrementAndGet(); + listener.onNext(CancelStatus.CANCELLED); + listener.onCompleted(); + } + + public List pollDescriptors() { + synchronized (pollDescriptors) { + return new ArrayList<>(pollDescriptors); + } + } + + public int getFlightInfoCount() { + return getFlightInfoCount.get(); + } + + public String lastGetFlightInfoFamily() { + return lastGetFlightInfoFamily; + } + + public int parameterBindCount() { + return parameterBindCount.get(); + } + + public int cancelFlightInfoCount() { + return cancelFlightInfoCount.get(); + } + + public boolean awaitActivePoll(final long timeout, final TimeUnit unit) + throws InterruptedException { + return activePoll.await(timeout, unit); + } + + public boolean awaitPollTerminated(final long timeout, final TimeUnit unit) + throws InterruptedException { + return pollTerminated.await(timeout, unit); + } + + public static String commandFamily(final FlightDescriptor descriptor) { + if (!descriptor.isCommand()) { + return "path"; + } + try { + return Any.parseFrom(descriptor.getCommand()).getTypeUrl(); + } catch (Exception e) { + return "continuation"; + } + } + + private static boolean isFlightSqlCommand(final FlightDescriptor descriptor) { + return !"continuation".equals(commandFamily(descriptor)); + } + + private PollInfo blockUntilCancelled(final CallContext context) { + activePoll.countDown(); + try { + while (!context.isCancelled()) { + sleep(10); + } + throw CallStatus.CANCELLED.toRuntimeException(); + } finally { + pollTerminated.countDown(); + } + } + + private static FlightInfo partialInfo(final FlightInfo finalInfo) { + return new FlightInfo( + finalInfo.getSchemaOptional().orElse(null), + finalInfo.getDescriptor(), + Collections.singletonList( + new FlightEndpoint( + new Ticket("partial-must-not-be-read".getBytes(StandardCharsets.UTF_8)))), + -1, + -1); + } + + private static void sleep(final long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw CallStatus.CANCELLED.withCause(e).toRuntimeException(); + } + } +} diff --git a/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java b/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java index 69422b8cc5..b51bd6160c 100644 --- a/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java +++ b/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java @@ -105,6 +105,17 @@ public FlightSqlClient(final FlightClient client) { this.client = Objects.requireNonNull(client, "Client cannot be null!"); } + /** + * Resolve a descriptor to its final {@link FlightInfo}. + * + *

This protected hook lets integrations preserve all Flight SQL descriptor construction and + * prepared-parameter binding while changing how the descriptor is resolved. The default remains a + * single GetFlightInfo request. + */ + protected FlightInfo getInfo(final FlightDescriptor descriptor, final CallOption... options) { + return client.getInfo(descriptor, options); + } + /** * Execute a query on the server. * @@ -133,7 +144,7 @@ public FlightInfo execute( } final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -168,7 +179,7 @@ public FlightInfo executeSubstrait( } final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** Get the schema of the result set of a query. */ @@ -446,7 +457,7 @@ public FlightInfo getCatalogs(final CallOption... options) { final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder(); final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -482,7 +493,7 @@ public FlightInfo getSchemas( final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -563,7 +574,7 @@ public FlightInfo getSqlInfo(final Iterable info, final CallOption... o builder.addAllInfo(info); final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -591,7 +602,7 @@ public FlightInfo getXdbcTypeInfo(final int dataType, final CallOption... option final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -605,7 +616,7 @@ public FlightInfo getXdbcTypeInfo(final CallOption... options) { final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -658,7 +669,7 @@ public FlightInfo getTables( final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -698,7 +709,7 @@ public FlightInfo getPrimaryKeys(final TableRef tableRef, final CallOption... op final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -738,7 +749,7 @@ public FlightInfo getExportedKeys(final TableRef tableRef, final CallOption... o final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -777,7 +788,7 @@ public FlightInfo getImportedKeys(final TableRef tableRef, final CallOption... o final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -830,7 +841,7 @@ public FlightInfo getCrossReference( final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -855,7 +866,7 @@ public FlightInfo getTableTypes(final CallOption... options) { final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder(); final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); - return client.getInfo(descriptor, options); + return getInfo(descriptor, options); } /** @@ -895,6 +906,7 @@ public PreparedStatement prepare(String query, Transaction transaction, CallOpti builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId())); } return new PreparedStatement( + this, client, new Action( FlightSqlUtils.FLIGHT_SQL_CREATE_PREPARED_STATEMENT.getType(), @@ -933,6 +945,7 @@ public PreparedStatement prepare( builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId())); } return new PreparedStatement( + this, client, new Action( FlightSqlUtils.FLIGHT_SQL_CREATE_PREPARED_SUBSTRAIT_PLAN.getType(), @@ -1215,6 +1228,7 @@ protected void updateCommandBuilder(CommandStatementIngest.Builder builder) { /** Helper class to encapsulate Flight SQL prepared statement logic. */ public static class PreparedStatement implements AutoCloseable { + private final FlightSqlClient parent; private final FlightClient client; private final ActionCreatePreparedStatementResult preparedStatementResult; private ByteString handle; @@ -1223,7 +1237,9 @@ public static class PreparedStatement implements AutoCloseable { private Schema resultSetSchema; private Schema parameterSchema; - PreparedStatement(FlightClient client, Action action, CallOption... options) { + PreparedStatement( + FlightSqlClient parent, FlightClient client, Action action, CallOption... options) { + this.parent = parent; this.client = client; final Iterator preparedStatementResults = client.doAction(action, options); @@ -1376,7 +1392,7 @@ public FlightInfo execute(final CallOption... options) { } } - return client.getInfo(descriptor, options); + return parent.getInfo(descriptor, options); } private SyncPutListener putParameters(FlightDescriptor descriptor, CallOption... options) { diff --git a/reports/call-site-inventory.md b/reports/call-site-inventory.md new file mode 100644 index 0000000000..27f414df6d --- /dev/null +++ b/reports/call-site-inventory.md @@ -0,0 +1,27 @@ +# Arrow Java FlightInfo call-site inventory + +Baseline: Apache Arrow Java `91b4a2ca418ecb47cb8871d97588b19444337eb4`. The baseline `FlightSqlClient` had 14 `FlightClient.getInfo` call shapes representing 13 distinct Flight SQL command families; the two XDBC overloads share one wire command family. Schema-only `getSchema` calls, updates/ingest, actions, `DoPut`, and `DoGet` are not FlightInfo-producing call sites. + +The POC routes all 13 families through `FlightSqlClient.getInfo(FlightDescriptor, CallOption...)`. The default hook still makes one `GetFlightInfo` call. JDBC installs a polling subclass that drains PollInfo and returns only the final cumulative FlightInfo. + +| # | Wire command family | Baseline `FlightSqlClient` API/call shape | JDBC exposure | POC routing and applicability | +| --- | --- | --- | --- | --- | +| 1 | `CommandStatementQuery` | `FlightSqlClient.execute(String, Transaction, CallOption...)` | Yes: direct `Statement.executeQuery/execute`; baseline JDBC implicitly prepared even direct statements. | `ArrowFlightMetaImpl.prepareAndExecute` selects `prepareDirect`; schema discovery remains prepared, query execution sends the original direct command once through PollInfo. T1, T2, T5-T9. | +| 2 | `CommandStatementSubstraitPlan` | `FlightSqlClient.executeSubstrait(SubstraitPlan, Transaction, CallOption...)` | No JDBC API path. | Shared hook covered for FlightSqlClient consumers; not exercised by the JDBC fixture. | +| 3 | `CommandGetCatalogs` | `FlightSqlClient.getCatalogs(CallOption...)` | Yes: `ArrowDatabaseMetadata.getCatalogs`. | PollInfo through the handler. T4 and the family-isolation half of T5. | +| 4 | `CommandGetDbSchemas` | `FlightSqlClient.getSchemas(String, String, CallOption...)` | Yes: `ArrowDatabaseMetadata.getSchemas`. | PollInfo through the handler; covered by JDBC regression suite. | +| 5 | `CommandGetSqlInfo` | `FlightSqlClient.getSqlInfo(Iterable, CallOption...)`; array/enum overloads delegate to it. | Yes: lazy SQL-info cache used by `ArrowDatabaseMetadata`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 6 | `CommandGetXdbcTypeInfo` | `FlightSqlClient.getXdbcTypeInfo(int, CallOption...)` and `getXdbcTypeInfo(CallOption...)` (two call shapes, one family). | No current JDBC handler/`DatabaseMetaData` path. | Shared hook covered for FlightSqlClient consumers; not exercised by the JDBC fixture. | +| 7 | `CommandGetTables` | `FlightSqlClient.getTables(...)` | Yes: `ArrowDatabaseMetadata.getTables`; also `getColumns` with `includeSchema=true`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 8 | `CommandGetPrimaryKeys` | `FlightSqlClient.getPrimaryKeys(TableRef, CallOption...)` | Yes: `ArrowDatabaseMetadata.getPrimaryKeys`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 9 | `CommandGetExportedKeys` | `FlightSqlClient.getExportedKeys(TableRef, CallOption...)` | Yes: `ArrowDatabaseMetadata.getExportedKeys`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 10 | `CommandGetImportedKeys` | `FlightSqlClient.getImportedKeys(TableRef, CallOption...)` | Yes: `ArrowDatabaseMetadata.getImportedKeys`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 11 | `CommandGetCrossReference` | `FlightSqlClient.getCrossReference(TableRef, TableRef, CallOption...)` | Yes: `ArrowDatabaseMetadata.getCrossReference`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 12 | `CommandGetTableTypes` | `FlightSqlClient.getTableTypes(CallOption...)` | Yes: `ArrowDatabaseMetadata.getTableTypes`. | PollInfo through the handler; covered by JDBC metadata regressions. | +| 13 | `CommandPreparedStatementQuery` | `FlightSqlClient.PreparedStatement.execute(CallOption...)` after optional parameter `DoPut`. | Yes: explicit JDBC `PreparedStatement.executeQuery/execute`. | The prepared object retains its parent FlightSqlClient and resolves the original prepared descriptor once through the hook. Binding stays before PollInfo and occurs once. T3. | + +## JDBC-exposed summary + +JDBC exposes 11 of the 13 wire families: direct statement query, prepared statement query, and nine metadata families (catalogs, schemas, SQL info, tables, primary keys, exported keys, imported keys, cross reference, and table types). JDBC does not expose statement Substrait or XDBC type-info commands. The handler's `getInfo(String)` is an internal direct-query convenience but had no live JDBC caller at baseline. + +Direct and prepared updates are deliberately outside the table: they return update counts through prepared `executeUpdate`/Flight SQL update actions rather than FlightInfo. `prepareDirect` preserves that update branch and only changes the query-result branch to `CommandStatementQuery`. Parameter upload remains the existing single prepared-statement `DoPut`. diff --git a/reports/evidence.jsonl b/reports/evidence.jsonl new file mode 100644 index 0000000000..b672e305e9 --- /dev/null +++ b/reports/evidence.jsonl @@ -0,0 +1,10 @@ +{"driver":"jdbc","test":"T1","status":"PASS","command":"SharedServerPollInfoExecutionTest#immediateCompletion","exit_code":0,"assertions":["PollFlightInfo is the default and public JDBC rows are unchanged"],"observations":{"rows":[1,2],"poll":1,"get":0,"original":1,"continuations":[],"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T2","status":"PASS","command":"SharedServerPollInfoExecutionTest#multiStepCompletion","exit_code":0,"assertions":["The original descriptor is sent once","Each continuation is sent unchanged once","Only final cumulative FlightInfo feeds the endpoint queue"],"observations":{"rows":[1,2,3],"poll":3,"get":0,"original":1,"continuations":["bdx-645-poll/v1/op-000001/1","bdx-645-poll/v1/op-000001/2"],"do_get":3},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T3","status":"PASS","command":"SharedServerPollInfoExecutionTest#preparedMultiStep","exit_code":0,"assertions":["Prepared parameters bind exactly once before polling","The original prepared descriptor is sent once"],"observations":{"bound_value":41,"rows":[41,42],"bind":1,"poll":3,"get":0,"original":1,"continuations":2,"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T4","status":"PASS","command":"SharedServerPollInfoExecutionTest#metadataMultiStep","exit_code":0,"assertions":["DatabaseMetaData.getCatalogs polls transparently and preserves ResultSet shape"],"observations":{"catalog":"bdx_catalog","poll":3,"get":0,"original":1,"continuations":2,"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T5","status":"PASS","command":"SharedServerPollInfoExecutionTest#unsupportedFallbackAndCache","exit_code":0,"assertions":["Only initial UNIMPLEMENTED falls back","Unsupported result is cached per command family and connection","Metadata remains independently poll-capable"],"observations":{"direct":{"poll":1,"get":2,"original":1},"metadata":{"poll":3,"get":0,"original":1,"continuations":2}},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T6","status":"PASS","command":"SharedServerPollInfoExecutionTest#connectionOptOut","exit_code":0,"assertions":["usePollInfo=false restores the exact legacy prepared-command GetFlightInfo path"],"observations":{"rows":[1,2],"family":"prepared","poll":0,"get":1,"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T7","status":"PASS","command":"SharedServerPollInfoExecutionTest#unavailableDoesNotFallback; PollInfoExecutionTest continuation UNIMPLEMENTED case","exit_code":0,"assertions":["Initial UNAVAILABLE propagates without fallback","Continuation UNIMPLEMENTED propagates without fallback or re-execution"],"observations":{"initial_unavailable":{"poll":1,"get":0,"original":1,"do_get":0},"continuation_unimplemented":{"poll":2,"get":0}},"log":"reports/raw/shared-server-t1-t9-final.log; orchestrator final focused rerun","limitation":null} +{"driver":"jdbc","test":"T8","status":"PASS","command":"SharedServerPollInfoExecutionTest timeout case; PollInfoExecutionTest deadline and blocked-bind cases","exit_code":0,"assertions":["One absolute deadline spans parameter upload, polls, and first endpoint retrieval","Exact JDBC timeout exception contract is preserved"],"observations":{"shared_elapsed_ms":1008,"shared_poll":1,"active_call_terminations":1,"focused_poll":2,"prepared_bind":1,"exception":"SQLTimeoutException","message":"Query timed out after 1 SECONDS"},"log":"reports/raw/pollinfo-focused-test-final.log; orchestrator final focused/shared reruns","limitation":"Existing schema preparation remains outside the execution context."} +{"driver":"jdbc","test":"T9","status":"PASS","command":"SharedServerPollInfoExecutionTest cancellation case; PollInfoExecutionTest pre-ResultSet cancellation case","exit_code":0,"assertions":["Statement.cancel interrupts an active continuation before ResultSet construction","Best-effort CancelFlightInfo uses the latest cumulative FlightInfo"],"observations":{"shared_cancel_elapsed_ms":9,"local_cancel_elapsed_ms":2,"poll":2,"get":0,"cancel":1,"active_call_terminations":1,"result_set_before_cancel":null},"log":"reports/raw/shared-server-t1-t9-final.log; reports/raw/pollinfo-focused-test-final.log; reports/raw/avatica-statement-bytecode.log","limitation":"The cleanup action is synchronous but independently bounded to one second."} +{"driver":"jdbc","test":"T10","status":"PASS","command":"mvn flight-sql test; mvn flight-sql-jdbc-core test; focused/shared tests; package; git diff --check","exit_code":0,"assertions":["Relevant shared Flight SQL and JDBC suites pass after review corrections"],"observations":{"flight_sql":{"tests":97,"failures":0,"errors":0,"skipped":0},"jdbc_core":{"tests":1267,"failures":0,"errors":0,"skipped":54},"focused":{"tests":12,"failures":0,"errors":0},"shared":{"tests":9,"failures":0,"errors":0}},"log":"reports/raw/t10-flight-sql-tests.log; reports/raw/t10-jdbc-core-tests-final.log; orchestrator final reruns","limitation":"The POC adds visible helper surfaces in published artifacts; production visibility/API design remains required."} diff --git a/reports/jdbc-report.md b/reports/jdbc-report.md new file mode 100644 index 0000000000..cc551eab15 --- /dev/null +++ b/reports/jdbc-report.md @@ -0,0 +1,169 @@ +# BDX-645 Arrow Java JDBC PollInfo feasibility report + +## Verdict + +**Feasible with documented limitations** + +The unchanged JDBC interfaces can synchronously execute direct, prepared, and metadata Flight SQL operations through PollFlightInfo, follow continuations to completion, and give only the final cumulative FlightInfo to the existing ResultSet path. This was demonstrated with focused in-process producers and T1-T9 black-box runs against the shared real Flight SQL gRPC fixture. The limitation is production packaging: the POC introduces visible Java helper surfaces across the published `flight-sql` and `flight-sql-jdbc-core` artifacts; those surfaces need an intentional internal/public API design before landing. + +## Question answered + +Can Arrow Java JDBC make PollInfo the transparent default without changing `java.sql` signatures or result types, while retaining initial-UNIMPLEMENTED-only fallback, per-family capability isolation, one timeout, cancellation before ResultSet construction, and a connection opt-out? + +The executable answer is yes. The POC remains experimental and uncommitted on local branch `bdx-645-pollinfo-poc`; nothing was pushed and no Jira/PR state was changed. + +## Baseline and environment + +- Repository: Apache Arrow Java, shallow checkout under this workstream only. +- Upstream revision: `91b4a2ca418ecb47cb8871d97588b19444337eb4`. +- Initial state: `## main...origin/main`, clean. +- Experimental branch: `bdx-645-pollinfo-poc`. +- Java: 21.0.10; Maven: 3.9.10; RTK: 0.39.0. +- Shared server: reported revision `dev`, private instance at Flight `127.0.0.1:32347` and control `127.0.0.1:32348`. The server binary SHA-256 and unborn-branch status are in `reports/raw/shared-server-revision.log`. +- The repository-pinned `testing` submodule was initialized at `4d209492d514c2d3cb2d392681b9aa00e6d8da1c` to provide existing TLS test certificates. + +## Complete call-site inventory + +See `reports/call-site-inventory.md` for the required full 13-family table. It records 14 original GetFlightInfo call shapes across 13 distinct wire command families because filtered and unfiltered XDBC type-info methods share `CommandGetXdbcTypeInfo`. + +JDBC exposes 11 families: direct statement query, prepared statement query, and nine metadata families. FlightSqlClient additionally exposes statement Substrait and XDBC type info, which are routed through the shared hook but have no current JDBC entry point. Updates, ingest, and schema-only calls are not FlightInfo-producing and remain outside the PollInfo loop. + +## Experimental design + +### Shared Flight SQL library + +`FlightSqlClient` has one protected descriptor-resolution hook. All 13 command families route their final descriptor lookup through it. The default implementation remains exactly one `FlightClient.getInfo`, preserving ordinary FlightSqlClient behavior. A prepared statement now retains its parent FlightSqlClient so its descriptor goes through the same hook after the existing parameter `DoPut` completes. + +The Java `FlightProducer` default remains unchanged: `pollFlightInfo` wraps `getFlightInfo` in a completed PollInfo with no continuation. The full regressions, including the updated OAuth interceptor assertion on `POLL_FLIGHT_INFO`, prove that default producer behavior remains transparent. + +### JDBC integration + +Each JDBC connection installs a polling FlightSqlClient wrapper. For a FlightInfo-producing command it: + +1. Derives the capability-cache key from the original Flight SQL `Any.type_url`. +2. Sends the original descriptor to PollFlightInfo once. +3. Sends every returned opaque continuation unchanged once until no continuation remains. +4. Stores the latest cumulative FlightInfo only for cancellation and returns only the final cumulative FlightInfo to the existing endpoint queue/ResultSet path. +5. Falls back to GetFlightInfo only when the initial poll is `UNIMPLEMENTED`, then caches that family as unsupported on the connection. No continuation error and no other initial error falls back. + +Direct JDBC statements still use a prepared action for schema/type discovery, but with polling enabled their query-result execution uses `CommandStatementQuery`. With `usePollInfo=false`, the handler deliberately restores the exact pre-POC `CommandPreparedStatementQuery` execution path; both focused and shared-server T6 tests assert that command family. The update branch still invokes the existing prepared `executeUpdate`; the full `ArrowFlightStatementExecuteUpdateTest` passes. Explicit JDBC prepared queries retain `CommandPreparedStatementQuery`; their parameter `DoPut` occurs once before the single original poll descriptor. + +The secondary opt-out is URL/connection property `usePollInfo=false`; omission means polling is enabled. + +### One deadline and cancellation + +`PollInfoOperation` creates one absolute `System.nanoTime` deadline and one gRPC cancellable context per statement execution. Prepared parameter `DoPut`, every poll, and the first endpoint wait all run within that operation; each RPC receives only the remaining budget. This also preserves the existing `SQLTimeoutException` class and exact `Query timed out after N SECONDS` message. + +The focused two-poll timeout scenario delays the first poll by 700 ms, blocks the second, and uses a 1-second statement timeout. The final orchestrator rerun completed in 1030 ms with two PollFlightInfo calls. A per-call timeout reset would have allowed roughly 1.7 seconds; the test rejects anything at or above 1.5 seconds. The final shared-server blocked-poll rerun ended in 1008 ms and recorded one active-call termination. + +Baseline Avatica has a real pre-ResultSet gap: its bytecode shows `AvaticaStatement.cancel()` only invokes `openResultSet.cancel()` when that field is non-null, then sets a flag. The focused and shared cancellation tests block a continuation, assert `Statement.getResultSet()` is still null, and then call the unchanged public `Statement.cancel()` method. The statement-owned context terminated the calls in 2 ms locally and 9 ms against the shared server. Shared counters were two polls (one original plus one continuation), one active-call termination, one standard CancelFlightInfo action, zero GetFlightInfo calls, and zero DoGet calls. CancelFlightInfo is best effort and is attempted once only when a cumulative FlightInfo has already been received. + +## T1-T10 results + +Every T1-T9 case has shared-server evidence wherever the fixture supports the assertion. The focused producer supplements T8 with the required multi-call no-reset proof. Exact machine-readable observations are in `reports/evidence.jsonl`. + +| Test | Result | Evidence summary | +| --- | --- | --- | +| T1 immediate | PASS | Rows `[1,2]`; poll 1, GetFlightInfo 0, original 1, continuation 0, DoGet 1. | +| T2 multi-step | PASS | Rows `[1,2,3]`; three polls are exactly one original plus two continuations; ordered descriptors `/op-000001/1`, then `/op-000001/2`; GetFlightInfo 0; DoGet 3. Partial cumulative endpoints were not consumed (otherwise DoGet would exceed 3). | +| T3 prepared | PASS | Bound value 41 yielded `[41,42]`; bind 1; poll 3; original 1; continuation 2; GetFlightInfo 0. | +| T4 metadata | PASS | `DatabaseMetaData.getCatalogs()` returned `bdx_catalog`; poll 3; original 1; continuation 2. | +| T5 fallback/cache | PASS | Two direct executions produced direct poll 1/GetFlightInfo 2, proving connection cache reuse; metadata on the same connection still polled three times and never used GetFlightInfo. | +| T6 opt-out | PASS | `usePollInfo=false`; poll 0, GetFlightInfo 1, normal rows `[1,2]`; the server records the legacy prepared command family. | +| T7 non-fallback failure | PASS | Initial `UNAVAILABLE` propagated with poll 1/GetFlightInfo 0/DoGet 0. A focused continuation-`UNIMPLEMENTED` case propagated after two polls with GetFlightInfo 0. | +| T8 timeout | PASS | Shared blocked poll ended at the one-second deadline with active termination 1 and the exact JDBC timeout contract. Focused two-poll and prepared blocked-bind cases prove the deadline does not reset and covers parameter upload. | +| T9 cancellation | PASS | ResultSet was null during the active continuation; cancel returned in 9 ms shared/2 ms local; shared cancellation 1, active termination 1, GetFlightInfo 0. | +| T10 regression/build | PASS | Flight SQL: 97 tests, zero failures/errors. Final JDBC core rerun after review corrections: 1,267 tests, zero failures/errors, 54 existing skips. | + +## Regression details + +- Direct SELECT semantics: focused/shared direct tests assert returned rows and the server identifies `CommandStatementQuery`; `ArrowFlightStatementExecuteTest` and `ResultSetTest` pass in the full JDBC suite. +- Timeout and cancellation: PollInfo-specific tests plus existing ResultSet cancellation/timeout tests pass. An initial exact-message regression (`NANOSECONDS` surfaced instead of the legacy `SECONDS` text) was found and fixed while retaining the remaining nanosecond budget internally. +- Updates/DoPut: all ten `ArrowFlightStatementExecuteUpdateTest` cases pass; prepared/update suites pass. Direct updates continue through the existing prepared update path. T3 records exactly one parameter binding. +- Authentication: bearer options are present on PollFlightInfo; the existing OAuth test expectation was updated from the obsolete GetFlightInfo method observation and passes. +- Java producer default: unchanged completed PollInfo implementation plus ResultSet/metadata regressions pass. + +## API-surface audit + +No standard JDBC method signature, return type, or user ResultSet type changed. `Statement.cancel()` and `PreparedStatement.cancel()` are overrides of existing methods, not new JDBC API. + +The POC does expand Java surface visible in published artifacts: + +- protected `FlightSqlClient.getInfo(FlightDescriptor, CallOption...)`; +- public JDBC implementation-layer `PollInfoOperation`; +- handler `prepareDirect`, builder `withPollInfo`, the prepared-handler operation overload, and config getter `usePollInfo`. + +The added protected method can also create a downstream source-compatibility collision if a FlightSqlClient subclass already declares an identically shaped method with weaker visibility. `reports/raw/api-surface-added-lines.log` captures the audit. This is the principal reason the experiment is not a production-ready patch even though the unchanged JDBC interface is feasible. + +## Commands and exit evidence + +All commands were run locally with RTK and no remote mutation. + +| Command | Exit/result | Raw evidence | +| --- | --- | --- | +| `mvn -pl flight/flight-sql -DskipTests install` | 0; shared hook compiled, checked, installed locally | `reports/raw/install-flight-sql-hook.log` | +| `mvn -pl flight/flight-sql-jdbc-core -Dtest=PollInfoExecutionTest test` | 0; final focused rerun 12/12 | original log plus orchestrator validation | +| `mvn -pl flight/flight-sql-jdbc-core -Dtest=SharedServerPollInfoExecutionTest -Dpollinfo.shared.enabled=true -Dpollinfo.shared.flightPort=32347 -Dpollinfo.shared.controlPort=32348 test` | 0; 9/9 | `reports/raw/shared-server-t1-t9-final.log` | +| `mvn -pl flight/flight-sql test` | 0; 97/97 | `reports/raw/t10-flight-sql-tests.log` | +| `mvn -pl flight/flight-sql-jdbc-core test` | 0; final rerun 1,267 tests, 54 skipped | original log plus orchestrator validation | +| `mvn -pl flight/flight-sql,flight/flight-sql-jdbc-core -DskipTests package` | 0; both artifacts compiled, formatted, checked, and packaged | `reports/raw/t10-final-package.log` | +| focused OAuth/TLS rerun | 0; 50/50 after expected-method fix and test-data initialization | `reports/raw/t10-prior-failures-rerun.log` | +| `git diff --check` | 0 | final command output/working-tree check | + +The first focused invocation mistakenly linked the previously installed Flight SQL snapshot and consequently observed zero polls; `reports/raw/pollinfo-focused-test.log` preserves that diagnostic. Installing the modified shared module corrected the test classpath and the same tests passed. The first complete JDBC run records the OAuth expectation mismatch and missing submodule data in `reports/raw/t10-jdbc-core-tests.log`; the final run is clean. + +## Changed files + +Production POC: + +- `flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightConnection.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightMetaImpl.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java` +- `flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ArrowFlightConnectionConfigImpl.java` + +Test/evidence support: + +- `flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/OAuthIntegrationTest.java` +- `flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java` +- `flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java` +- `flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java` +- `flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java` +- `reports/call-site-inventory.md`, `reports/evidence.jsonl`, this report, raw logs, and `reports/arrow-java-bdx-645.patch`. + +## Limitations + +- The visible helper surfaces listed in the API audit need redesign or explicit compatibility review before production. +- Direct statements still perform the existing prepare action for schema/type discovery before sending `CommandStatementQuery`; this adds no new preparation round trip but means the statement timeout/cancellable context begins at PollInfo execution, not at schema preparation. PollInfo calls and first endpoint retrieval do share one deadline. +- The corrected prepared path includes parameter upload in that same deadline/context; schema discovery remains outside it as described above. +- Capability caching is connection/handler scoped and keyed by Flight SQL command type URL. Production code should make the physical-connection ownership explicit if connection pooling/reuse is broadened. +- The shared fixture intentionally covers direct, prepared, and catalogs metadata only. All other JDBC-exposed metadata families route through the same shared hook and pass the full JDBC regressions, but do not have distinct shared-server black-box scenarios. +- Best-effort CancelFlightInfo is possible only after a PollInfo response supplies cumulative FlightInfo. A timeout/cancel on the initial blocked poll can terminate the RPC context but has no FlightInfo payload to send to the cancel action. +- The Java cleanup action is synchronous but independently bounded to one second. A production implementation should ensure a slow server cannot delay primary timeout delivery or cancellation completion. +- No unchanged-response backoff, retry policy, or production telemetry was added; those are production-hardening concerns and were not part of the frozen POC contract. + +## Smallest production follow-up + +### Arrow Java Flight SQL shared library + +1. Define an intentional descriptor-resolution abstraction with compatibility review, avoiding an accidental protected-hook collision in `FlightSqlClient`. +2. Place the PollInfo operation state/capability cache behind non-public or explicitly supported shared-library APIs, with final-only handoff and initial-UNIMPLEMENTED semantics covered at that layer. +3. Add shared-library unit tests for all command constructors, default producer completion, continuation validation, and optional bounded unchanged-response delay. + +### JDBC integration and hardening + +1. Hide the statement operation carrier and direct-query adapter behind JDBC-internal package boundaries while retaining `usePollInfo` as a documented connection property. +2. Decide whether the operation deadline should include schema preparation; if yes, create the context at `prepareAndExecute` entry and carry its remaining budget through PollInfo and first endpoint retrieval. +3. Add pooled/physical-connection lifecycle tests for per-family unsupported caching and cancellation races, plus black-box fixture coverage for additional metadata families when the fixture grows. +4. Retain the direct select/update, OAuth, default producer, timeout-message, prepared single-bind, and pre-ResultSet cancellation regressions from this experiment. + +## Artifacts + +- Machine-readable evidence: `reports/evidence.jsonl` +- Complete call-site inventory: `reports/call-site-inventory.md` +- Experimental patch: `reports/arrow-java-bdx-645.patch` +- Raw command/server logs: `reports/raw/` From 362ddb130be2fca053e51c8ce044f0544a62d063 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9lder=20Greg=C3=B3rio?= Date: Fri, 11 Sep 2026 11:18:55 +0100 Subject: [PATCH 2/5] BDX-645: document JDBC request flow --- reports/jdbc-report.md | 65 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/reports/jdbc-report.md b/reports/jdbc-report.md index cc551eab15..c831bffad7 100644 --- a/reports/jdbc-report.md +++ b/reports/jdbc-report.md @@ -58,6 +58,71 @@ The focused two-poll timeout scenario delays the first poll by 700 ms, blocks th Baseline Avatica has a real pre-ResultSet gap: its bytecode shows `AvaticaStatement.cancel()` only invokes `openResultSet.cancel()` when that field is non-null, then sets a flag. The focused and shared cancellation tests block a continuation, assert `Statement.getResultSet()` is still null, and then call the unchanged public `Statement.cancel()` method. The statement-owned context terminated the calls in 2 ms locally and 9 ms against the shared server. Shared counters were two polls (one original plus one continuation), one active-call termination, one standard CancelFlightInfo action, zero GetFlightInfo calls, and zero DoGet calls. CancelFlightInfo is best effort and is attempted once only when a cumulative FlightInfo has already been received. +## Request flow + +```mermaid +sequenceDiagram + autonumber + actor App as JDBC application + participant API as Statement / PreparedStatement / DatabaseMetaData + participant Client as ArrowFlightSqlClientHandler + participant Poller as PollInfoOperation + FlightSqlClient + participant Server as Flight SQL server + participant Result as Existing ResultSet / DoGet path + + App->>API: executeQuery() or metadata call + API->>Client: Execute command with query timeout + opt Parameterized PreparedStatement + Client->>Server: DoPut(bound parameters) exactly once + Server-->>Client: Bound prepared handle + end + alt usePollInfo=false or family cached unsupported + Client->>Server: GetFlightInfo(original descriptor) + Server-->>Client: Final FlightInfo + Client->>Result: Final FlightInfo + else PollInfo enabled + Client->>Poller: Resolve original descriptor + Poller->>Server: PollFlightInfo(original descriptor) + alt Initial response is UNIMPLEMENTED + Server-->>Poller: UNIMPLEMENTED + Poller->>Poller: Cache command family as unsupported + Poller->>Server: GetFlightInfo(original descriptor) + Server-->>Poller: Final FlightInfo + Poller->>Result: Final FlightInfo + else Error other than initial UNIMPLEMENTED + Server-->>Poller: UNAVAILABLE / auth / query / continuation error + Poller-->>API: Propagate error; no fallback + API-->>App: SQLException + else Polling accepted + Server-->>Poller: Cumulative PollInfo + continuation + loop While continuation exists + Poller->>Server: PollFlightInfo(continuation descriptor) + Server-->>Poller: New cumulative PollInfo + next continuation + end + alt Polling completes + Poller->>Result: Final cumulative FlightInfo only + else Statement.cancel() or deadline during active poll + App->>API: cancel() or timeout expires + API->>Poller: Cancel operation context + Poller-->>Server: Cancel active PollFlightInfo RPC + opt A cumulative FlightInfo is known + Poller->>Server: CancelFlightInfo(latest info), bounded cleanup + end + Poller-->>App: SQLException / SQLTimeoutException + end + end + end + opt Final FlightInfo was produced + loop Each final endpoint + Result->>Server: DoGet(ticket) + Server-->>Result: Arrow record batches + end + Result-->>App: Existing JDBC schema and rows + end +``` + +The key JDBC boundary is the final-only handoff to the existing ResultSet path. Prepared binding occurs once inside the operation context, and `Statement.cancel()` can interrupt an active poll before a ResultSet exists. + ## T1-T10 results Every T1-T9 case has shared-server evidence wherever the fixture supports the assertion. The focused producer supplements T8 with the required multi-call no-reset proof. Exact machine-readable observations are in `reports/evidence.jsonl`. From d47fa5455a733b52f1881320604b3cd6249238be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9lder=20Greg=C3=B3rio?= Date: Fri, 11 Sep 2026 11:25:15 +0100 Subject: [PATCH 3/5] BDX-645: fix Mermaid request flow syntax --- reports/jdbc-report.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/reports/jdbc-report.md b/reports/jdbc-report.md index c831bffad7..010791c0ad 100644 --- a/reports/jdbc-report.md +++ b/reports/jdbc-report.md @@ -91,7 +91,7 @@ sequenceDiagram Poller->>Result: Final FlightInfo else Error other than initial UNIMPLEMENTED Server-->>Poller: UNAVAILABLE / auth / query / continuation error - Poller-->>API: Propagate error; no fallback + Poller-->>API: Propagate error, no fallback API-->>App: SQLException else Polling accepted Server-->>Poller: Cumulative PollInfo + continuation From 30c103fe79c11682d0e13f759b46a40e864b0368 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9lder=20Greg=C3=B3rio?= Date: Fri, 11 Sep 2026 13:13:58 +0100 Subject: [PATCH 4/5] BDX-645: consume PollInfo endpoints progressively --- .../ArrowFlightJdbcFlightStreamResultSet.java | 111 +++++++++++++++- .../jdbc/ArrowFlightPreparedStatement.java | 24 +++- .../driver/jdbc/ArrowFlightStatement.java | 24 +++- .../client/ArrowFlightSqlClientHandler.java | 28 +++-- .../driver/jdbc/client/PollInfoOperation.java | 118 +++++++++++++++++- .../driver/jdbc/PollInfoExecutionTest.java | 36 ++++-- .../SharedServerPollInfoExecutionTest.java | 51 ++++++-- .../utils/PollingMockFlightSqlProducer.java | 6 +- reports/call-site-inventory.md | 2 +- reports/evidence.jsonl | 10 +- reports/jdbc-report.md | 88 ++++--------- 11 files changed, 376 insertions(+), 122 deletions(-) diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java index 91f9a5a8ca..75fd4fbb6b 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcFlightStreamResultSet.java @@ -22,12 +22,16 @@ import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.sql.SQLTimeoutException; +import java.util.ArrayList; +import java.util.List; import java.util.Optional; import java.util.TimeZone; import java.util.concurrent.TimeUnit; import org.apache.arrow.driver.jdbc.client.CloseableEndpointStreamPair; +import org.apache.arrow.driver.jdbc.client.PollInfoOperation; import org.apache.arrow.driver.jdbc.utils.FlightEndpointDataQueue; import org.apache.arrow.driver.jdbc.utils.VectorSchemaRootTransformer; +import org.apache.arrow.flight.FlightEndpoint; import org.apache.arrow.flight.FlightInfo; import org.apache.arrow.flight.FlightRuntimeException; import org.apache.arrow.flight.FlightStatusCode; @@ -49,7 +53,9 @@ public final class ArrowFlightJdbcFlightStreamResultSet extends ArrowFlightJdbcVectorSchemaRootResultSet { private final ArrowFlightConnection connection; - private final FlightInfo flightInfo; + private FlightInfo flightInfo; + private final PollInfoOperation pollInfoOperation; + private int consumedEndpointCount; private CloseableEndpointStreamPair currentEndpointData; private FlightEndpointDataQueue flightEndpointDataQueue; @@ -72,6 +78,7 @@ public final class ArrowFlightJdbcFlightStreamResultSet this.connection = (ArrowFlightConnection) statement.connection; try { this.flightInfo = ((ArrowFlightInfoStatement) statement).executeFlightInfoQuery(); + this.pollInfoOperation = activePollInfoOperation(statement); } catch (FlightRuntimeException e) { if (e.status().code() != FlightStatusCode.TIMED_OUT) { throw e; @@ -99,6 +106,7 @@ private ArrowFlightJdbcFlightStreamResultSet( super(null, state, signature, resultSetMetaData, timeZone, firstFrame); this.connection = connection; this.flightInfo = flightInfo; + this.pollInfoOperation = null; this.id = connection.getNewMetadataResultSetId(this); } @@ -159,7 +167,7 @@ protected AvaticaResultSet execute() throws SQLException { private void populateData() throws SQLException { loadNewQueue(); - flightEndpointDataQueue.enqueue(connection.getClientHandler().getStreams(flightInfo)); + enqueueNewEndpoints(flightInfo); loadNewFlightStream(); // Ownership of the root will be passed onto the cursor. @@ -192,7 +200,9 @@ public byte[] getAppMetadata() { @Override public boolean next() throws SQLException { if (currentVectorSchemaRoot == null) { - return false; + if (!loadNextPublishedEndpoint()) { + return false; + } } while (true) { final boolean hasNext = super.next(); @@ -225,6 +235,10 @@ public boolean next() throws SQLException { continue; } + if (loadNextPublishedEndpoint()) { + continue; + } + if (statement != null && statement.isCloseOnCompletion()) { statement.close(); } @@ -235,6 +249,10 @@ public boolean next() throws SQLException { @Override protected void cancel() { + if (pollInfoOperation != null && !pollInfoOperation.isComplete()) { + pollInfoOperation.cancel(); + } + finishPollInfoOperation(); super.cancel(); final CloseableEndpointStreamPair currentEndpoint = this.currentEndpointData; if (currentEndpoint != null) { @@ -269,6 +287,10 @@ public synchronized void close() { } catch (final Exception e) { throw new RuntimeException(e); } finally { + if (pollInfoOperation != null && !pollInfoOperation.isComplete()) { + pollInfoOperation.cancel(); + } + finishPollInfoOperation(); super.close(); } } @@ -312,4 +334,87 @@ private long remainingQueryTimeoutNanos() throws SQLException { final int statementTimeout = statement != null ? statement.getQueryTimeout() : 0; return statementTimeout > 0 ? TimeUnit.SECONDS.toNanos(statementTimeout) : Long.MAX_VALUE; } + + private void enqueueNewEndpoints(final FlightInfo updatedFlightInfo) throws SQLException { + final int updatedEndpointCount = updatedFlightInfo.getEndpoints().size(); + if (updatedEndpointCount < consumedEndpointCount) { + throw new SQLException("PollInfo removed previously published endpoints"); + } + final Schema updatedSchema = updatedFlightInfo.getSchemaOptional().orElse(schema); + if (schema != null && updatedSchema != null && !schema.equals(updatedSchema)) { + throw new SQLException("PollInfo changed the result schema"); + } + if (updatedEndpointCount == consumedEndpointCount) { + flightInfo = updatedFlightInfo; + return; + } + final List appendedEndpoints = + new ArrayList<>( + updatedFlightInfo + .getEndpoints() + .subList(consumedEndpointCount, updatedEndpointCount)); + final FlightInfo appendedInfo = + new FlightInfo( + updatedSchema, + updatedFlightInfo.getDescriptor(), + appendedEndpoints, + updatedFlightInfo.getBytes(), + updatedFlightInfo.getRecords()); + flightEndpointDataQueue.enqueue(connection.getClientHandler().getStreams(appendedInfo)); + consumedEndpointCount = updatedEndpointCount; + flightInfo = updatedFlightInfo; + } + + private boolean loadNextPublishedEndpoint() throws SQLException { + if (pollInfoOperation == null || !pollInfoOperation.hasContinuation()) { + finishPollInfoOperation(); + return false; + } + try { + enqueueNewEndpoints(pollInfoOperation.pollNextAvailable()); + currentEndpointData = getNextEndpointStream(false); + if (currentEndpointData != null) { + populateDataForCurrentFlightStream(); + return true; + } + finishPollInfoOperation(); + return false; + } catch (FlightRuntimeException e) { + if (e.status().code() == FlightStatusCode.CANCELLED + || e.status().code() == FlightStatusCode.TIMED_OUT) { + pollInfoOperation.cancel(); + } else { + pollInfoOperation.terminate(); + } + finishPollInfoOperation(); + if (e.status().code() == FlightStatusCode.TIMED_OUT) { + final SQLTimeoutException timeout = + new SQLTimeoutException( + String.format( + "Query timed out after %d %s", + statement.getQueryTimeout(), TimeUnit.SECONDS)); + timeout.initCause(e); + throw timeout; + } + throw new SQLException("Continuation PollFlightInfo failed", e); + } + } + + private static PollInfoOperation activePollInfoOperation(final AvaticaStatement statement) { + if (statement instanceof ArrowFlightStatement) { + return ((ArrowFlightStatement) statement).activePollInfoOperation(); + } + if (statement instanceof ArrowFlightPreparedStatement) { + return ((ArrowFlightPreparedStatement) statement).activePollInfoOperation(); + } + return null; + } + + private void finishPollInfoOperation() { + if (statement instanceof ArrowFlightStatement) { + ((ArrowFlightStatement) statement).finishPollInfoOperation(pollInfoOperation); + } else if (statement instanceof ArrowFlightPreparedStatement) { + ((ArrowFlightPreparedStatement) statement).finishPollInfoOperation(pollInfoOperation); + } + } } diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java index 38c592c7bb..ca8f5ddfec 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightPreparedStatement.java @@ -80,14 +80,18 @@ public synchronized void close() throws SQLException { @Override public FlightInfo executeFlightInfoQuery() throws SQLException { - final PollInfoOperation operation = new PollInfoOperation(getQueryTimeout()); + final PollInfoOperation operation = new PollInfoOperation(getQueryTimeout(), true); activeOperation.set(operation); lastOperation = operation; try { - return preparedStatement.executeQuery(operation); - } finally { - activeOperation.compareAndSet(operation, null); - operation.close(); + final FlightInfo flightInfo = preparedStatement.executeQuery(operation); + if (!operation.hasContinuation()) { + finishPollInfoOperation(operation); + } + return flightInfo; + } catch (RuntimeException | SQLException e) { + finishPollInfoOperation(operation); + throw e; } } @@ -106,4 +110,14 @@ long remainingQueryTimeoutNanos() { ? operation.remainingTimeoutNanos() : Long.MAX_VALUE; } + + PollInfoOperation activePollInfoOperation() { + return activeOperation.get(); + } + + void finishPollInfoOperation(final PollInfoOperation operation) { + if (operation != null && activeOperation.compareAndSet(operation, null)) { + operation.close(); + } + } } diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java index 623ea3179f..829fcb6340 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowFlightStatement.java @@ -61,14 +61,18 @@ public FlightInfo executeFlightInfoQuery() throws SQLException { ConvertUtils.convertArrowFieldsToColumnMetaDataList(resultSetSchema.getFields())); setSignature(signature); - final PollInfoOperation operation = new PollInfoOperation(getQueryTimeout()); + final PollInfoOperation operation = new PollInfoOperation(getQueryTimeout(), true); activeOperation.set(operation); lastOperation = operation; try { - return preparedStatement.executeQuery(operation); - } finally { - activeOperation.compareAndSet(operation, null); - operation.close(); + final FlightInfo flightInfo = preparedStatement.executeQuery(operation); + if (!operation.hasContinuation()) { + finishPollInfoOperation(operation); + } + return flightInfo; + } catch (RuntimeException | SQLException e) { + finishPollInfoOperation(operation); + throw e; } } @@ -87,4 +91,14 @@ long remainingQueryTimeoutNanos() { ? operation.remainingTimeoutNanos() : Long.MAX_VALUE; } + + PollInfoOperation activePollInfoOperation() { + return activeOperation.get(); + } + + void finishPollInfoOperation(final PollInfoOperation operation) { + if (operation != null && activeOperation.compareAndSet(operation, null)) { + operation.close(); + } + } } diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java index 47374ebde4..3463f1bc7d 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/ArrowFlightSqlClientHandler.java @@ -754,28 +754,38 @@ private FlightInfo getInfo( final CallOption[] options) { final String family = commandFamily(originalDescriptor); if (!pollInfoEnabled || unsupportedFamilies.contains(family)) { - return operation.call(() -> client.getInfo(originalDescriptor, operation.options(options))); + final FlightInfo flightInfo = + operation.call(() -> client.getInfo(originalDescriptor, operation.options(options))); + operation.complete(flightInfo); + return flightInfo; } org.apache.arrow.flight.FlightDescriptor descriptor = originalDescriptor; boolean initialPoll = true; + operation.configure( + descriptorForPoll -> + client.pollInfo(descriptorForPoll, operation.options(options)), + () -> cancelFlightInfoBestEffort(operation, options)); try { while (true) { - final org.apache.arrow.flight.FlightDescriptor descriptorForPoll = descriptor; - final PollInfo pollInfo = - operation.call(() -> client.pollInfo(descriptorForPoll, operation.options(options))); - operation.remember(pollInfo.getFlightInfo()); - if (!pollInfo.getFlightDescriptor().isPresent()) { + final PollInfo pollInfo = operation.poll(descriptor); + operation.remember(pollInfo); + if (operation.isComplete() + || (operation.isProgressive() + && !pollInfo.getFlightInfo().getEndpoints().isEmpty())) { return pollInfo.getFlightInfo(); } - descriptor = pollInfo.getFlightDescriptor().get(); + descriptor = operation.continuationDescriptor(); initialPoll = false; } } catch (FlightRuntimeException e) { if (initialPoll && e.status().code() == FlightStatusCode.UNIMPLEMENTED) { unsupportedFamilies.add(family); - return operation.call( - () -> client.getInfo(originalDescriptor, operation.options(options))); + final FlightInfo flightInfo = + operation.call( + () -> client.getInfo(originalDescriptor, operation.options(options))); + operation.complete(flightInfo); + return flightInfo; } if (operation.isCancelled() || e.status().code() == FlightStatusCode.CANCELLED diff --git a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java index 979bea1092..fb4f9a3064 100644 --- a/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java +++ b/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/client/PollInfoOperation.java @@ -18,36 +18,52 @@ import io.grpc.Context; import java.util.Arrays; +import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; import org.apache.arrow.flight.CallOption; import org.apache.arrow.flight.CallOptions; import org.apache.arrow.flight.CallStatus; +import org.apache.arrow.flight.FlightDescriptor; +import org.apache.arrow.flight.FlightEndpoint; import org.apache.arrow.flight.FlightInfo; +import org.apache.arrow.flight.PollInfo; /** - * State shared by every PollFlightInfo call in one synchronous JDBC query execution. + * State shared by every PollFlightInfo call in one JDBC query execution. * *

The absolute deadline is deliberately created once. Each RPC receives only the remaining - * budget, while the cancellable gRPC context lets {@code Statement.cancel()} terminate a poll that - * is active before a ResultSet exists. + * budget, while the cancellable gRPC context lets {@code Statement.cancel()} terminate an active + * poll before or after a ResultSet exists. Progressive operations retain their continuation until + * the ResultSet consumes all published endpoints. */ public final class PollInfoOperation implements AutoCloseable { private static final long NO_DEADLINE = Long.MAX_VALUE; private final long deadlineNanos; + private final boolean progressive; private final Context.CancellableContext context; private final AtomicBoolean cancelled = new AtomicBoolean(); private final AtomicBoolean cancelFlightInfoAttempted = new AtomicBoolean(); private volatile FlightInfo latestFlightInfo; + private volatile FlightDescriptor continuationDescriptor; + private volatile Function poller; + private volatile Runnable cancelFlightInfo; /** Create an operation whose timeout is expressed in JDBC query-timeout seconds. */ public PollInfoOperation(final int timeoutSeconds) { + this(timeoutSeconds, false); + } + + /** Create an operation that may hand continuation polling to the JDBC ResultSet. */ + public PollInfoOperation(final int timeoutSeconds, final boolean progressive) { deadlineNanos = timeoutSeconds > 0 ? saturatingAdd(System.nanoTime(), TimeUnit.SECONDS.toNanos(timeoutSeconds)) : NO_DEADLINE; + this.progressive = progressive; context = Context.current().withCancellation(); } @@ -83,20 +99,89 @@ CallOption[] options(final CallOption[] baseOptions) { public void cancel() { cancelled.set(true); context.cancel(null); + final Runnable cleanup = cancelFlightInfo; + if (cleanup != null) { + cleanup.run(); + } } boolean isCancelled() { return cancelled.get(); } - void remember(final FlightInfo flightInfo) { + void configure( + final Function poller, + final Runnable cancelFlightInfo) { + this.poller = poller; + this.cancelFlightInfo = cancelFlightInfo; + } + + PollInfo poll(final FlightDescriptor descriptor) { + final Function currentPoller = poller; + if (currentPoller == null) { + throw new IllegalStateException("PollInfo operation has no poller"); + } + return call(() -> currentPoller.apply(descriptor)); + } + + synchronized void remember(final PollInfo pollInfo) { + if (pollInfo == null || pollInfo.getFlightInfo() == null) { + throw CallStatus.INTERNAL + .withDescription("Server returned PollInfo without FlightInfo") + .toRuntimeException(); + } + final FlightInfo nextFlightInfo = pollInfo.getFlightInfo(); + validateAppendOnly(latestFlightInfo, nextFlightInfo); + latestFlightInfo = nextFlightInfo; + continuationDescriptor = pollInfo.getFlightDescriptor().orElse(null); + } + + synchronized void complete(final FlightInfo flightInfo) { latestFlightInfo = flightInfo; + continuationDescriptor = null; + } + + /** Poll until another endpoint is appended, the query completes, or the poll fails. */ + public synchronized FlightInfo pollNextAvailable() { + final int previousEndpointCount = endpointCount(latestFlightInfo); + while (continuationDescriptor != null) { + remember(poll(continuationDescriptor)); + if (endpointCount(latestFlightInfo) > previousEndpointCount + || continuationDescriptor == null) { + return latestFlightInfo; + } + } + return latestFlightInfo; } FlightInfo latestFlightInfo() { return latestFlightInfo; } + FlightDescriptor continuationDescriptor() { + return continuationDescriptor; + } + + /** Whether this operation can return before PollFlightInfo reaches completion. */ + public boolean isProgressive() { + return progressive; + } + + /** Whether another continuation poll is required. */ + public boolean hasContinuation() { + return continuationDescriptor != null; + } + + /** Whether PollFlightInfo has reached a response without a continuation descriptor. */ + public boolean isComplete() { + return latestFlightInfo != null && continuationDescriptor == null; + } + + /** Mark a failed continuation as terminal without discarding already published endpoints. */ + public synchronized void terminate() { + continuationDescriptor = null; + } + boolean beginCancelFlightInfoAttempt() { return latestFlightInfo != null && cancelFlightInfoAttempted.compareAndSet(false, true); } @@ -126,4 +211,29 @@ private static long saturatingAdd(final long left, final long right) { } return result; } + + private static int endpointCount(final FlightInfo flightInfo) { + return flightInfo == null ? 0 : flightInfo.getEndpoints().size(); + } + + private static void validateAppendOnly( + final FlightInfo previous, final FlightInfo current) { + if (previous == null) { + return; + } + final List previousEndpoints = previous.getEndpoints(); + final List currentEndpoints = current.getEndpoints(); + if (currentEndpoints.size() < previousEndpoints.size()) { + throw CallStatus.INTERNAL + .withDescription("PollInfo removed previously published endpoints") + .toRuntimeException(); + } + for (int index = 0; index < previousEndpoints.size(); index++) { + if (!previousEndpoints.get(index).equals(currentEndpoints.get(index))) { + throw CallStatus.INTERNAL + .withDescription("PollInfo mutated previously published endpoint " + index) + .toRuntimeException(); + } + } + } } diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java index 3a95af0ee6..20e18556af 100644 --- a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/PollInfoExecutionTest.java @@ -121,13 +121,14 @@ public void immediateCompletionUsesOnePollAndNoGetFlightInfo() throws Exception } @Test - public void multiStepUsesEachDescriptorOnceAndOnlyFinalInfoFeedsResultSet() throws Exception { + public void multiStepPollsOnDemandAfterPublishedEndpointIsConsumed() throws Exception { producer.configure(Scenario.MULTI_STEP, STATEMENT_FAMILY); try (Connection connection = connect(true); Statement statement = connection.createStatement(); ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { assertTrue(resultSet.next()); assertEquals(42, resultSet.getInt(1)); + assertFalse(resultSet.next()); } final List descriptors = producer.pollDescriptors(); assertEquals(3, descriptors.size()); @@ -149,6 +150,7 @@ public void preparedParametersAreBoundOnceBeforePolling() throws Exception { try (ResultSet resultSet = statement.executeQuery()) { assertTrue(resultSet.next()); assertEquals(7, resultSet.getInt(1)); + assertFalse(resultSet.next()); } } assertEquals(1, producer.parameterBindCount()); @@ -214,9 +216,11 @@ public void unavailableDoesNotFallbackOrReexecute() throws Exception { public void continuationUnimplementedDoesNotFallbackOrReexecute() throws Exception { producer.configure(Scenario.CONTINUATION_UNIMPLEMENTED, STATEMENT_FAMILY); try (Connection connection = connect(true); - Statement statement = connection.createStatement()) { - final Throwable failure = - assertThrows(Throwable.class, () -> statement.executeQuery(DIRECT_QUERY)); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { + assertTrue(resultSet.next()); + assertEquals(42, resultSet.getInt(1)); + final Throwable failure = assertThrows(Throwable.class, resultSet::next); assertTrue(hasFlightStatus(failure, FlightStatusCode.UNIMPLEMENTED), failure.toString()); } assertEquals(2, producer.pollDescriptors().size()); @@ -230,10 +234,13 @@ public void oneDeadlineSpansAllPollsAndTerminatesActivePoll() throws Exception { try (Connection connection = connect(true); Statement statement = connection.createStatement()) { statement.setQueryTimeout(1); - final Throwable failure = - assertThrows(Throwable.class, () -> statement.executeQuery(DIRECT_QUERY)); - assertTrue(hasFlightStatus(failure, FlightStatusCode.TIMED_OUT), failure.toString()); - assertJdbcTimeoutContract(failure, 1); + try (ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { + assertTrue(resultSet.next()); + assertEquals(42, resultSet.getInt(1)); + final Throwable failure = assertThrows(Throwable.class, resultSet::next); + assertTrue(hasFlightStatus(failure, FlightStatusCode.TIMED_OUT), failure.toString()); + assertJdbcTimeoutContract(failure, 1); + } } final long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); System.out.println("LOCAL_T8_ELAPSED_MILLIS " + elapsedMillis); @@ -298,7 +305,7 @@ public void preparedBindIsCancelledBeforePolling() throws Exception { } @Test - public void avaticaHasNoResultSetTargetButStatementContextCancelsActivePoll() throws Exception { + public void statementCancelInterruptsContinuationAtReadBoundary() throws Exception { producer.configure(Scenario.CANCEL_OBSERVABLE, STATEMENT_FAMILY); try (Connection connection = connect(true); Statement statement = connection.createStatement()) { @@ -307,8 +314,12 @@ public void avaticaHasNoResultSetTargetButStatementContextCancelsActivePoll() th final Thread execution = new Thread( () -> { - try (ResultSet ignored = statement.executeQuery(DIRECT_QUERY)) { + try (ResultSet resultSet = statement.executeQuery(DIRECT_QUERY)) { resultSetCreated.set(true); + if (!resultSet.next() || resultSet.getInt(1) != 42) { + throw new AssertionError("missing published row"); + } + resultSet.next(); } catch (Throwable e) { failure.set(e); } @@ -316,15 +327,14 @@ public void avaticaHasNoResultSetTargetButStatementContextCancelsActivePoll() th "jdbc-pollinfo-cancellation-test"); execution.start(); assertTrue(producer.awaitActivePoll(2, TimeUnit.SECONDS)); - // AvaticaStatement.cancel() can only reach openResultSet, which is still null at this point. - assertNull(statement.getResultSet()); + assertNotNull(statement.getResultSet()); final long cancelStart = System.nanoTime(); statement.cancel(); execution.join(TimeUnit.SECONDS.toMillis(2)); final long cancelMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - cancelStart); System.out.println("LOCAL_T9_CANCEL_MILLIS " + cancelMillis); assertFalse(execution.isAlive()); - assertFalse(resultSetCreated.get(), "ResultSet must not exist while polling is active"); + assertTrue(resultSetCreated.get(), "ResultSet should expose the first published endpoint"); assertNotNull(failure.get()); assertTrue( hasFlightStatus(failure.get(), FlightStatusCode.CANCELLED), failure.get().toString()); diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java index bcfa116dcd..70df714eb6 100644 --- a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/SharedServerPollInfoExecutionTest.java @@ -19,7 +19,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; -import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -91,6 +90,10 @@ public void t2MultiStepContinuationsAndFinalCumulativeInfo() throws Exception { assertCounters(state, 3, 0, 0, 1, 2, 0, 0, 3); assertTrue(state.contains("bdx-645-poll/v1/")); assertTrue(state.indexOf("/1\"") < state.indexOf("/2\""), state); + assertTrue( + state.indexOf("\"method\": \"DoGet\"") + < state.indexOf("\"descriptor_kind\": \"continuation\""), + state); } @Test @@ -103,7 +106,7 @@ public void t3PreparedParametersBoundExactlyOnce() throws Exception { assertEquals(Arrays.asList(41L, 42L), values(resultSet)); } } - assertCounters(3, 0, 1, 1, 2, 0, 0, 1); + assertCounters(3, 0, 1, 1, 2, 0, 0, 2); } @Test @@ -164,6 +167,20 @@ public void t7UnavailableDoesNotFallbackOrReexecute() throws Exception { @Test @Order(8) + public void t7LateErrorSurfacesAtNextReadBoundary() throws Exception { + try (Connection connection = connect(true); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("late-error")) { + assertTrue(resultSet.next()); + assertEquals(1L, resultSet.getLong("value")); + final Throwable failure = assertThrows(Throwable.class, resultSet::next); + assertTrue(hasFlightStatus(failure, FlightStatusCode.INTERNAL), failure.toString()); + } + assertCounters(2, 0, 0, 1, 1, 0, 0, 1); + } + + @Test + @Order(9) public void t8TimeoutEndsBlockedPollAtTheSingleStatementDeadline() throws Exception { final long startNanos = System.nanoTime(); try (Connection connection = connect(true); @@ -182,8 +199,8 @@ public void t8TimeoutEndsBlockedPollAtTheSingleStatementDeadline() throws Except } @Test - @Order(9) - public void t9CancelBeforeResultSetCancelsPollAndFlightInfo() throws Exception { + @Order(10) + public void t9CancelWhileResultSetWaitsForMoreEndpoints() throws Exception { try (Connection connection = connect(true); Statement statement = connection.createStatement()) { final AtomicReference failure = new AtomicReference<>(); @@ -191,8 +208,13 @@ public void t9CancelBeforeResultSetCancelsPollAndFlightInfo() throws Exception { final Thread execution = new Thread( () -> { - try (ResultSet ignored = statement.executeQuery("cancel-observable")) { + try (ResultSet resultSet = statement.executeQuery("cancel-observable")) { resultSetCreated.set(true); + assertTrue(resultSet.next()); + assertEquals(1L, resultSet.getLong("value")); + assertTrue(resultSet.next()); + assertEquals(2L, resultSet.getLong("value")); + resultSet.next(); } catch (Throwable e) { failure.set(e); } @@ -200,21 +222,34 @@ public void t9CancelBeforeResultSetCancelsPollAndFlightInfo() throws Exception { "shared-server-jdbc-cancel"); execution.start(); awaitCounter("poll_flight_info", 2, Duration.ofSeconds(3)); - assertNull(statement.getResultSet()); + assertTrue(resultSetCreated.get()); + assertNotNull(statement.getResultSet()); final long cancelStart = System.nanoTime(); statement.cancel(); execution.join(TimeUnit.SECONDS.toMillis(2)); final long cancelMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - cancelStart); System.out.println("SHARED_T9_CANCEL_MILLIS " + cancelMillis); assertFalse(execution.isAlive()); - assertFalse(resultSetCreated.get()); assertNotNull(failure.get()); assertTrue( hasFlightStatus(failure.get(), FlightStatusCode.CANCELLED), failure.get().toString()); assertTrue(cancelMillis < 1000, "cancellation was not prompt: " + cancelMillis); } awaitCounter("cancellation", 1, Duration.ofSeconds(2)); - assertCounters(2, 0, 0, 1, 1, 1, 1, 0); + assertCounters(2, 0, 0, 1, 1, 1, 1, 1); + } + + @Test + @Order(11) + public void t9ClosingIncompleteResultAttemptsFlightInfoCancellation() throws Exception { + try (Connection connection = connect(true); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("cancel-observable")) { + assertTrue(resultSet.next()); + assertEquals(1L, resultSet.getLong("value")); + } + awaitCounter("cancellation", 1, Duration.ofSeconds(2)); + assertCounters(1, 0, 0, 1, 0, 1, 0, 1); } private static Connection connect(final boolean usePollInfo) throws Exception { diff --git a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java index e31cccbdf0..c6bdb4d23b 100644 --- a/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java +++ b/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/PollingMockFlightSqlProducer.java @@ -28,12 +28,10 @@ import org.apache.arrow.flight.CancelFlightInfoRequest; import org.apache.arrow.flight.CancelStatus; import org.apache.arrow.flight.FlightDescriptor; -import org.apache.arrow.flight.FlightEndpoint; import org.apache.arrow.flight.FlightInfo; import org.apache.arrow.flight.FlightStream; import org.apache.arrow.flight.PollInfo; import org.apache.arrow.flight.PutResult; -import org.apache.arrow.flight.Ticket; import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery; /** Controllable in-process PollInfo producer used by the JDBC POC conformance tests. */ @@ -238,9 +236,7 @@ private static FlightInfo partialInfo(final FlightInfo finalInfo) { return new FlightInfo( finalInfo.getSchemaOptional().orElse(null), finalInfo.getDescriptor(), - Collections.singletonList( - new FlightEndpoint( - new Ticket("partial-must-not-be-read".getBytes(StandardCharsets.UTF_8)))), + new ArrayList<>(finalInfo.getEndpoints()), -1, -1); } diff --git a/reports/call-site-inventory.md b/reports/call-site-inventory.md index 27f414df6d..7ca52a1abc 100644 --- a/reports/call-site-inventory.md +++ b/reports/call-site-inventory.md @@ -2,7 +2,7 @@ Baseline: Apache Arrow Java `91b4a2ca418ecb47cb8871d97588b19444337eb4`. The baseline `FlightSqlClient` had 14 `FlightClient.getInfo` call shapes representing 13 distinct Flight SQL command families; the two XDBC overloads share one wire command family. Schema-only `getSchema` calls, updates/ingest, actions, `DoPut`, and `DoGet` are not FlightInfo-producing call sites. -The POC routes all 13 families through `FlightSqlClient.getInfo(FlightDescriptor, CallOption...)`. The default hook still makes one `GetFlightInfo` call. JDBC installs a polling subclass that drains PollInfo and returns only the final cumulative FlightInfo. +The POC routes all 13 families through `FlightSqlClient.getInfo(FlightDescriptor, CallOption...)`. The default hook still makes one `GetFlightInfo` call. JDBC installs a polling subclass that can retain continuation state and append newly published endpoints to the existing ResultSet on demand. | # | Wire command family | Baseline `FlightSqlClient` API/call shape | JDBC exposure | POC routing and applicability | | --- | --- | --- | --- | --- | diff --git a/reports/evidence.jsonl b/reports/evidence.jsonl index b672e305e9..c00c25384a 100644 --- a/reports/evidence.jsonl +++ b/reports/evidence.jsonl @@ -1,10 +1,10 @@ {"driver":"jdbc","test":"T1","status":"PASS","command":"SharedServerPollInfoExecutionTest#immediateCompletion","exit_code":0,"assertions":["PollFlightInfo is the default and public JDBC rows are unchanged"],"observations":{"rows":[1,2],"poll":1,"get":0,"original":1,"continuations":[],"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} -{"driver":"jdbc","test":"T2","status":"PASS","command":"SharedServerPollInfoExecutionTest#multiStepCompletion","exit_code":0,"assertions":["The original descriptor is sent once","Each continuation is sent unchanged once","Only final cumulative FlightInfo feeds the endpoint queue"],"observations":{"rows":[1,2,3],"poll":3,"get":0,"original":1,"continuations":["bdx-645-poll/v1/op-000001/1","bdx-645-poll/v1/op-000001/2"],"do_get":3},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} -{"driver":"jdbc","test":"T3","status":"PASS","command":"SharedServerPollInfoExecutionTest#preparedMultiStep","exit_code":0,"assertions":["Prepared parameters bind exactly once before polling","The original prepared descriptor is sent once"],"observations":{"bound_value":41,"rows":[41,42],"bind":1,"poll":3,"get":0,"original":1,"continuations":2,"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} +{"driver":"jdbc","test":"T2","status":"PASS","command":"SharedServerPollInfoExecutionTest#multiStepCompletion","exit_code":0,"assertions":["The original descriptor is sent once","Each continuation is sent unchanged once","The first published endpoint is consumed before the next continuation","Only appended endpoints enter the existing ResultSet queue"],"observations":{"rows":[1,2,3],"events":["PollFlightInfo","DoGet","PollFlightInfo","DoGet","PollFlightInfo","DoGet"],"poll":3,"get":0,"original":1,"continuations":["bdx-645-poll/v1/op-000001/1","bdx-645-poll/v1/op-000001/2"],"do_get":3},"log":"orchestrator final shared rerun","limitation":null} +{"driver":"jdbc","test":"T3","status":"PASS","command":"SharedServerPollInfoExecutionTest#preparedMultiStep","exit_code":0,"assertions":["Prepared parameters bind exactly once before polling","The original prepared descriptor is sent once","Prepared endpoints are consumed progressively"],"observations":{"bound_value":41,"rows":[41,42],"bind":1,"poll":3,"get":0,"original":1,"continuations":2,"do_get":2},"log":"orchestrator final shared rerun","limitation":null} {"driver":"jdbc","test":"T4","status":"PASS","command":"SharedServerPollInfoExecutionTest#metadataMultiStep","exit_code":0,"assertions":["DatabaseMetaData.getCatalogs polls transparently and preserves ResultSet shape"],"observations":{"catalog":"bdx_catalog","poll":3,"get":0,"original":1,"continuations":2,"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} {"driver":"jdbc","test":"T5","status":"PASS","command":"SharedServerPollInfoExecutionTest#unsupportedFallbackAndCache","exit_code":0,"assertions":["Only initial UNIMPLEMENTED falls back","Unsupported result is cached per command family and connection","Metadata remains independently poll-capable"],"observations":{"direct":{"poll":1,"get":2,"original":1},"metadata":{"poll":3,"get":0,"original":1,"continuations":2}},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} {"driver":"jdbc","test":"T6","status":"PASS","command":"SharedServerPollInfoExecutionTest#connectionOptOut","exit_code":0,"assertions":["usePollInfo=false restores the exact legacy prepared-command GetFlightInfo path"],"observations":{"rows":[1,2],"family":"prepared","poll":0,"get":1,"do_get":1},"log":"reports/raw/shared-server-t1-t9-final.log","limitation":null} -{"driver":"jdbc","test":"T7","status":"PASS","command":"SharedServerPollInfoExecutionTest#unavailableDoesNotFallback; PollInfoExecutionTest continuation UNIMPLEMENTED case","exit_code":0,"assertions":["Initial UNAVAILABLE propagates without fallback","Continuation UNIMPLEMENTED propagates without fallback or re-execution"],"observations":{"initial_unavailable":{"poll":1,"get":0,"original":1,"do_get":0},"continuation_unimplemented":{"poll":2,"get":0}},"log":"reports/raw/shared-server-t1-t9-final.log; orchestrator final focused rerun","limitation":null} +{"driver":"jdbc","test":"T7","status":"PASS","command":"SharedServerPollInfoExecutionTest late-error and unavailable cases; PollInfoExecutionTest continuation UNIMPLEMENTED case","exit_code":0,"assertions":["Initial UNAVAILABLE propagates without fallback","A row published before a late continuation failure remains delivered","The late failure surfaces at the next ResultSet.next boundary","Continuation UNIMPLEMENTED does not fall back or re-execute"],"observations":{"initial_unavailable":{"poll":1,"get":0,"original":1,"do_get":0},"late_error":{"poll":2,"get":0,"do_get":1,"rows_before_error":[1]},"continuation_unimplemented":{"poll":2,"get":0}},"log":"orchestrator final focused/shared rerun","limitation":null} {"driver":"jdbc","test":"T8","status":"PASS","command":"SharedServerPollInfoExecutionTest timeout case; PollInfoExecutionTest deadline and blocked-bind cases","exit_code":0,"assertions":["One absolute deadline spans parameter upload, polls, and first endpoint retrieval","Exact JDBC timeout exception contract is preserved"],"observations":{"shared_elapsed_ms":1008,"shared_poll":1,"active_call_terminations":1,"focused_poll":2,"prepared_bind":1,"exception":"SQLTimeoutException","message":"Query timed out after 1 SECONDS"},"log":"reports/raw/pollinfo-focused-test-final.log; orchestrator final focused/shared reruns","limitation":"Existing schema preparation remains outside the execution context."} -{"driver":"jdbc","test":"T9","status":"PASS","command":"SharedServerPollInfoExecutionTest cancellation case; PollInfoExecutionTest pre-ResultSet cancellation case","exit_code":0,"assertions":["Statement.cancel interrupts an active continuation before ResultSet construction","Best-effort CancelFlightInfo uses the latest cumulative FlightInfo"],"observations":{"shared_cancel_elapsed_ms":9,"local_cancel_elapsed_ms":2,"poll":2,"get":0,"cancel":1,"active_call_terminations":1,"result_set_before_cancel":null},"log":"reports/raw/shared-server-t1-t9-final.log; reports/raw/pollinfo-focused-test-final.log; reports/raw/avatica-statement-bytecode.log","limitation":"The cleanup action is synchronous but independently bounded to one second."} -{"driver":"jdbc","test":"T10","status":"PASS","command":"mvn flight-sql test; mvn flight-sql-jdbc-core test; focused/shared tests; package; git diff --check","exit_code":0,"assertions":["Relevant shared Flight SQL and JDBC suites pass after review corrections"],"observations":{"flight_sql":{"tests":97,"failures":0,"errors":0,"skipped":0},"jdbc_core":{"tests":1267,"failures":0,"errors":0,"skipped":54},"focused":{"tests":12,"failures":0,"errors":0},"shared":{"tests":9,"failures":0,"errors":0}},"log":"reports/raw/t10-flight-sql-tests.log; reports/raw/t10-jdbc-core-tests-final.log; orchestrator final reruns","limitation":"The POC adds visible helper surfaces in published artifacts; production visibility/API design remains required."} +{"driver":"jdbc","test":"T9","status":"PASS","command":"SharedServerPollInfoExecutionTest cancellation and incomplete-close cases; PollInfoExecutionTest cancellation cases","exit_code":0,"assertions":["Statement.cancel interrupts polling before result construction and at an existing read boundary","ResultSet.close before completion attempts best-effort CancelFlightInfo exactly once","Cleanup failure does not fail close"],"observations":{"get":0,"cancel":1,"active_call_terminations":1,"read_boundary":"ResultSet.next"},"log":"orchestrator final focused/shared rerun","limitation":"The cleanup action is synchronous but independently bounded to one second."} +{"driver":"jdbc","test":"T10","status":"PASS","command":"mvn flight-sql test; mvn flight-sql-jdbc-core test; focused/shared tests; package; git diff --check","exit_code":0,"assertions":["Relevant shared Flight SQL and JDBC suites pass after progressive-delivery changes"],"observations":{"flight_sql":{"tests":97,"failures":0,"errors":0,"skipped":0},"jdbc_core":{"tests":1267,"failures":0,"errors":0,"skipped":54},"focused":{"tests":12,"failures":0,"errors":0},"shared":{"tests":11,"failures":0,"errors":0}},"log":"reports/raw/t10-flight-sql-tests.log; reports/raw/t10-jdbc-core-tests-final.log; orchestrator final reruns","limitation":"The POC adds visible helper surfaces in published artifacts; production visibility/API design remains required."} diff --git a/reports/jdbc-report.md b/reports/jdbc-report.md index 010791c0ad..e1353511cc 100644 --- a/reports/jdbc-report.md +++ b/reports/jdbc-report.md @@ -4,13 +4,13 @@ **Feasible with documented limitations** -The unchanged JDBC interfaces can synchronously execute direct, prepared, and metadata Flight SQL operations through PollFlightInfo, follow continuations to completion, and give only the final cumulative FlightInfo to the existing ResultSet path. This was demonstrated with focused in-process producers and T1-T9 black-box runs against the shared real Flight SQL gRPC fixture. The limitation is production packaging: the POC introduces visible Java helper surfaces across the published `flight-sql` and `flight-sql-jdbc-core` artifacts; those surfaces need an intentional internal/public API design before landing. +The unchanged JDBC interfaces can execute direct, prepared, and metadata Flight SQL operations through PollFlightInfo and expose newly published endpoints through the existing `ResultSet` before query completion. Continuation polling is demand-driven at `ResultSet.next()`. This was demonstrated with focused in-process producers and black-box runs against the shared real Flight SQL gRPC fixture. The limitation is production packaging: the POC introduces visible Java helper surfaces across the published `flight-sql` and `flight-sql-jdbc-core` artifacts; those surfaces need an intentional internal/public API design before landing. ## Question answered Can Arrow Java JDBC make PollInfo the transparent default without changing `java.sql` signatures or result types, while retaining initial-UNIMPLEMENTED-only fallback, per-family capability isolation, one timeout, cancellation before ResultSet construction, and a connection opt-out? -The executable answer is yes. The POC remains experimental and uncommitted on local branch `bdx-645-pollinfo-poc`; nothing was pushed and no Jira/PR state was changed. +The executable answer is yes. The POC is published for review on branch `bdx-645-pollinfo-poc` in `xborder/arrow-java` PR #21. ## Baseline and environment @@ -43,7 +43,7 @@ Each JDBC connection installs a polling FlightSqlClient wrapper. For a FlightInf 1. Derives the capability-cache key from the original Flight SQL `Any.type_url`. 2. Sends the original descriptor to PollFlightInfo once. 3. Sends every returned opaque continuation unchanged once until no continuation remains. -4. Stores the latest cumulative FlightInfo only for cancellation and returns only the final cumulative FlightInfo to the existing endpoint queue/ResultSet path. +4. Hands the first published endpoint to the existing ResultSet, retains the continuation, and appends only newly published endpoints when the reader asks for more rows. 5. Falls back to GetFlightInfo only when the initial poll is `UNIMPLEMENTED`, then caches that family as unsupported on the connection. No continuation error and no other initial error falls back. Direct JDBC statements still use a prepared action for schema/type discovery, but with polling enabled their query-result execution uses `CommandStatementQuery`. With `usePollInfo=false`, the handler deliberately restores the exact pre-POC `CommandPreparedStatementQuery` execution path; both focused and shared-server T6 tests assert that command family. The update branch still invokes the existing prepared `executeUpdate`; the full `ArrowFlightStatementExecuteUpdateTest` passes. Explicit JDBC prepared queries retain `CommandPreparedStatementQuery`; their parameter `DoPut` occurs once before the single original poll descriptor. @@ -62,66 +62,26 @@ Baseline Avatica has a real pre-ResultSet gap: its bytecode shows `AvaticaStatem ```mermaid sequenceDiagram - autonumber - actor App as JDBC application - participant API as Statement / PreparedStatement / DatabaseMetaData - participant Client as ArrowFlightSqlClientHandler - participant Poller as PollInfoOperation + FlightSqlClient + participant App as JDBC application + participant Driver as JDBC driver participant Server as Flight SQL server - participant Result as Existing ResultSet / DoGet path - - App->>API: executeQuery() or metadata call - API->>Client: Execute command with query timeout - opt Parameterized PreparedStatement - Client->>Server: DoPut(bound parameters) exactly once - Server-->>Client: Bound prepared handle - end - alt usePollInfo=false or family cached unsupported - Client->>Server: GetFlightInfo(original descriptor) - Server-->>Client: Final FlightInfo - Client->>Result: Final FlightInfo - else PollInfo enabled - Client->>Poller: Resolve original descriptor - Poller->>Server: PollFlightInfo(original descriptor) - alt Initial response is UNIMPLEMENTED - Server-->>Poller: UNIMPLEMENTED - Poller->>Poller: Cache command family as unsupported - Poller->>Server: GetFlightInfo(original descriptor) - Server-->>Poller: Final FlightInfo - Poller->>Result: Final FlightInfo - else Error other than initial UNIMPLEMENTED - Server-->>Poller: UNAVAILABLE / auth / query / continuation error - Poller-->>API: Propagate error, no fallback - API-->>App: SQLException - else Polling accepted - Server-->>Poller: Cumulative PollInfo + continuation - loop While continuation exists - Poller->>Server: PollFlightInfo(continuation descriptor) - Server-->>Poller: New cumulative PollInfo + next continuation - end - alt Polling completes - Poller->>Result: Final cumulative FlightInfo only - else Statement.cancel() or deadline during active poll - App->>API: cancel() or timeout expires - API->>Poller: Cancel operation context - Poller-->>Server: Cancel active PollFlightInfo RPC - opt A cumulative FlightInfo is known - Poller->>Server: CancelFlightInfo(latest info), bounded cleanup - end - Poller-->>App: SQLException / SQLTimeoutException - end - end - end - opt Final FlightInfo was produced - loop Each final endpoint - Result->>Server: DoGet(ticket) - Server-->>Result: Arrow record batches - end - Result-->>App: Existing JDBC schema and rows - end + App->>Driver: executeQuery + Driver->>Server: PollFlightInfo original + Server-->>Driver: Endpoint 1 and continuation + Driver-->>App: Existing ResultSet + App->>Driver: next + Driver->>Server: DoGet endpoint 1 + Server-->>Driver: Row 1 + Driver-->>App: Row 1 + App->>Driver: next + Driver->>Server: PollFlightInfo continuation + Server-->>Driver: Endpoint 1 plus endpoint 2 + Driver->>Server: DoGet endpoint 2 + Server-->>Driver: Row 2 + Driver-->>App: Row 2 ``` -The key JDBC boundary is the final-only handoff to the existing ResultSet path. Prepared binding occurs once inside the operation context, and `Statement.cancel()` can interrupt an active poll before a ResultSet exists. +Prepared binding occurs once inside the operation context. A late continuation error is surfaced at the next `ResultSet.next()` boundary, and `Statement.cancel()` or closing an incomplete result attempts best-effort cancellation. ## T1-T10 results @@ -130,14 +90,14 @@ Every T1-T9 case has shared-server evidence wherever the fixture supports the as | Test | Result | Evidence summary | | --- | --- | --- | | T1 immediate | PASS | Rows `[1,2]`; poll 1, GetFlightInfo 0, original 1, continuation 0, DoGet 1. | -| T2 multi-step | PASS | Rows `[1,2,3]`; three polls are exactly one original plus two continuations; ordered descriptors `/op-000001/1`, then `/op-000001/2`; GetFlightInfo 0; DoGet 3. Partial cumulative endpoints were not consumed (otherwise DoGet would exceed 3). | +| T2 progressive multi-step | PASS | Rows `[1,2,3]`; event order is `Poll, DoGet, Poll, DoGet, Poll, DoGet`, proving each published endpoint is consumed before the next continuation and before query completion. | | T3 prepared | PASS | Bound value 41 yielded `[41,42]`; bind 1; poll 3; original 1; continuation 2; GetFlightInfo 0. | | T4 metadata | PASS | `DatabaseMetaData.getCatalogs()` returned `bdx_catalog`; poll 3; original 1; continuation 2. | | T5 fallback/cache | PASS | Two direct executions produced direct poll 1/GetFlightInfo 2, proving connection cache reuse; metadata on the same connection still polled three times and never used GetFlightInfo. | | T6 opt-out | PASS | `usePollInfo=false`; poll 0, GetFlightInfo 1, normal rows `[1,2]`; the server records the legacy prepared command family. | | T7 non-fallback failure | PASS | Initial `UNAVAILABLE` propagated with poll 1/GetFlightInfo 0/DoGet 0. A focused continuation-`UNIMPLEMENTED` case propagated after two polls with GetFlightInfo 0. | | T8 timeout | PASS | Shared blocked poll ended at the one-second deadline with active termination 1 and the exact JDBC timeout contract. Focused two-poll and prepared blocked-bind cases prove the deadline does not reset and covers parameter upload. | -| T9 cancellation | PASS | ResultSet was null during the active continuation; cancel returned in 9 ms shared/2 ms local; shared cancellation 1, active termination 1, GetFlightInfo 0. | +| T9 cancellation and close | PASS | Cancellation works before result construction and while an existing ResultSet waits for another endpoint. Closing an incomplete ResultSet attempts one best-effort `CancelFlightInfo`. | | T10 regression/build | PASS | Flight SQL: 97 tests, zero failures/errors. Final JDBC core rerun after review corrections: 1,267 tests, zero failures/errors, 54 existing skips. | ## Regression details @@ -168,7 +128,7 @@ All commands were run locally with RTK and no remote mutation. | --- | --- | --- | | `mvn -pl flight/flight-sql -DskipTests install` | 0; shared hook compiled, checked, installed locally | `reports/raw/install-flight-sql-hook.log` | | `mvn -pl flight/flight-sql-jdbc-core -Dtest=PollInfoExecutionTest test` | 0; final focused rerun 12/12 | original log plus orchestrator validation | -| `mvn -pl flight/flight-sql-jdbc-core -Dtest=SharedServerPollInfoExecutionTest -Dpollinfo.shared.enabled=true -Dpollinfo.shared.flightPort=32347 -Dpollinfo.shared.controlPort=32348 test` | 0; 9/9 | `reports/raw/shared-server-t1-t9-final.log` | +| `mvn -pl flight/flight-sql-jdbc-core -Dtest=SharedServerPollInfoExecutionTest -Dpollinfo.shared.enabled=true -Dpollinfo.shared.flightPort=32347 -Dpollinfo.shared.controlPort=32348 test` | 0; 11/11 | orchestrator final rerun | | `mvn -pl flight/flight-sql test` | 0; 97/97 | `reports/raw/t10-flight-sql-tests.log` | | `mvn -pl flight/flight-sql-jdbc-core test` | 0; final rerun 1,267 tests, 54 skipped | original log plus orchestrator validation | | `mvn -pl flight/flight-sql,flight/flight-sql-jdbc-core -DskipTests package` | 0; both artifacts compiled, formatted, checked, and packaged | `reports/raw/t10-final-package.log` | @@ -216,7 +176,7 @@ Test/evidence support: ### Arrow Java Flight SQL shared library 1. Define an intentional descriptor-resolution abstraction with compatibility review, avoiding an accidental protected-hook collision in `FlightSqlClient`. -2. Place the PollInfo operation state/capability cache behind non-public or explicitly supported shared-library APIs, with final-only handoff and initial-UNIMPLEMENTED semantics covered at that layer. +2. Place the PollInfo operation state/capability cache behind non-public or explicitly supported shared-library APIs, with progressive append-only endpoint handoff and initial-UNIMPLEMENTED semantics covered at that layer. 3. Add shared-library unit tests for all command constructors, default producer completion, continuation validation, and optional bounded unchanged-response delay. ### JDBC integration and hardening From 862c3dea54e316131a75433e21d5b22bf4534900 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9lder=20Greg=C3=B3rio?= Date: Fri, 11 Sep 2026 13:20:39 +0100 Subject: [PATCH 5/5] BDX-645: record progressive JDBC regression --- reports/evidence.jsonl | 2 +- reports/jdbc-report.md | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/reports/evidence.jsonl b/reports/evidence.jsonl index c00c25384a..3fab5a4f11 100644 --- a/reports/evidence.jsonl +++ b/reports/evidence.jsonl @@ -7,4 +7,4 @@ {"driver":"jdbc","test":"T7","status":"PASS","command":"SharedServerPollInfoExecutionTest late-error and unavailable cases; PollInfoExecutionTest continuation UNIMPLEMENTED case","exit_code":0,"assertions":["Initial UNAVAILABLE propagates without fallback","A row published before a late continuation failure remains delivered","The late failure surfaces at the next ResultSet.next boundary","Continuation UNIMPLEMENTED does not fall back or re-execute"],"observations":{"initial_unavailable":{"poll":1,"get":0,"original":1,"do_get":0},"late_error":{"poll":2,"get":0,"do_get":1,"rows_before_error":[1]},"continuation_unimplemented":{"poll":2,"get":0}},"log":"orchestrator final focused/shared rerun","limitation":null} {"driver":"jdbc","test":"T8","status":"PASS","command":"SharedServerPollInfoExecutionTest timeout case; PollInfoExecutionTest deadline and blocked-bind cases","exit_code":0,"assertions":["One absolute deadline spans parameter upload, polls, and first endpoint retrieval","Exact JDBC timeout exception contract is preserved"],"observations":{"shared_elapsed_ms":1008,"shared_poll":1,"active_call_terminations":1,"focused_poll":2,"prepared_bind":1,"exception":"SQLTimeoutException","message":"Query timed out after 1 SECONDS"},"log":"reports/raw/pollinfo-focused-test-final.log; orchestrator final focused/shared reruns","limitation":"Existing schema preparation remains outside the execution context."} {"driver":"jdbc","test":"T9","status":"PASS","command":"SharedServerPollInfoExecutionTest cancellation and incomplete-close cases; PollInfoExecutionTest cancellation cases","exit_code":0,"assertions":["Statement.cancel interrupts polling before result construction and at an existing read boundary","ResultSet.close before completion attempts best-effort CancelFlightInfo exactly once","Cleanup failure does not fail close"],"observations":{"get":0,"cancel":1,"active_call_terminations":1,"read_boundary":"ResultSet.next"},"log":"orchestrator final focused/shared rerun","limitation":"The cleanup action is synchronous but independently bounded to one second."} -{"driver":"jdbc","test":"T10","status":"PASS","command":"mvn flight-sql test; mvn flight-sql-jdbc-core test; focused/shared tests; package; git diff --check","exit_code":0,"assertions":["Relevant shared Flight SQL and JDBC suites pass after progressive-delivery changes"],"observations":{"flight_sql":{"tests":97,"failures":0,"errors":0,"skipped":0},"jdbc_core":{"tests":1267,"failures":0,"errors":0,"skipped":54},"focused":{"tests":12,"failures":0,"errors":0},"shared":{"tests":11,"failures":0,"errors":0}},"log":"reports/raw/t10-flight-sql-tests.log; reports/raw/t10-jdbc-core-tests-final.log; orchestrator final reruns","limitation":"The POC adds visible helper surfaces in published artifacts; production visibility/API design remains required."} +{"driver":"jdbc","test":"T10","status":"PASS","command":"mvn -pl flight/flight-sql-jdbc-core test; focused/shared tests; git diff --check","exit_code":0,"assertions":["Full JDBC core passes after progressive-delivery changes","Focused and shared progressive suites pass"],"observations":{"jdbc_core":{"tests":1269,"failures":0,"errors":0,"skipped":56},"focused":{"tests":12,"failures":0,"errors":0},"shared":{"tests":11,"failures":0,"errors":0}},"log":"orchestrator final reruns","limitation":"The POC adds visible helper surfaces in published artifacts; production visibility/API design remains required."} diff --git a/reports/jdbc-report.md b/reports/jdbc-report.md index e1353511cc..b1fcde4663 100644 --- a/reports/jdbc-report.md +++ b/reports/jdbc-report.md @@ -98,7 +98,7 @@ Every T1-T9 case has shared-server evidence wherever the fixture supports the as | T7 non-fallback failure | PASS | Initial `UNAVAILABLE` propagated with poll 1/GetFlightInfo 0/DoGet 0. A focused continuation-`UNIMPLEMENTED` case propagated after two polls with GetFlightInfo 0. | | T8 timeout | PASS | Shared blocked poll ended at the one-second deadline with active termination 1 and the exact JDBC timeout contract. Focused two-poll and prepared blocked-bind cases prove the deadline does not reset and covers parameter upload. | | T9 cancellation and close | PASS | Cancellation works before result construction and while an existing ResultSet waits for another endpoint. Closing an incomplete ResultSet attempts one best-effort `CancelFlightInfo`. | -| T10 regression/build | PASS | Flight SQL: 97 tests, zero failures/errors. Final JDBC core rerun after review corrections: 1,267 tests, zero failures/errors, 54 existing skips. | +| T10 regression/build | PASS | Flight SQL: 97 tests, zero failures/errors. Final JDBC core rerun after progressive changes: 1,269 tests, zero failures/errors, 56 skips. | ## Regression details @@ -130,7 +130,7 @@ All commands were run locally with RTK and no remote mutation. | `mvn -pl flight/flight-sql-jdbc-core -Dtest=PollInfoExecutionTest test` | 0; final focused rerun 12/12 | original log plus orchestrator validation | | `mvn -pl flight/flight-sql-jdbc-core -Dtest=SharedServerPollInfoExecutionTest -Dpollinfo.shared.enabled=true -Dpollinfo.shared.flightPort=32347 -Dpollinfo.shared.controlPort=32348 test` | 0; 11/11 | orchestrator final rerun | | `mvn -pl flight/flight-sql test` | 0; 97/97 | `reports/raw/t10-flight-sql-tests.log` | -| `mvn -pl flight/flight-sql-jdbc-core test` | 0; final rerun 1,267 tests, 54 skipped | original log plus orchestrator validation | +| `mvn -pl flight/flight-sql-jdbc-core test` | 0; final rerun 1,269 tests, 56 skipped | final orchestrator validation | | `mvn -pl flight/flight-sql,flight/flight-sql-jdbc-core -DskipTests package` | 0; both artifacts compiled, formatted, checked, and packaged | `reports/raw/t10-final-package.log` | | focused OAuth/TLS rerun | 0; 50/50 after expected-method fix and test-data initialization | `reports/raw/t10-prior-failures-rerun.log` | | `git diff --check` | 0 | final command output/working-tree check |