You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "eramitmittal (via GitHub)" <gi...@apache.org> on 2024/08/04 17:58:21 UTC
[PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
eramitmittal opened a new pull request, #43551:
URL: https://github.com/apache/arrow/pull/43551
Please look at #38255 for details on this functionality. Support for GO and C++ was added as part of #38385.
This pull request is to add the required support for Java.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2287423722
Revision: 9d3a98126af086c2a889e596bb06b76e66de9c8e
Submitted crossbow builds: [ursacomputing/crossbow @ actions-ffe9da94a6](https://github.com/ursacomputing/crossbow/branches/all?query=actions-ffe9da94a6)
|Task|Status|
|----|------|
|java-jars|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940160/job/28736162739)|
|test-conda-python-3.11-spark-master|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940152/job/28736161778)|
|verify-rc-source-java-linux-almalinux-8-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940161/job/28736161973)|
|verify-rc-source-java-linux-conda-latest-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940123/job/28736161714)|
|verify-rc-source-java-linux-ubuntu-20.04-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940149/job/28736161721)|
|verify-rc-source-java-linux-ubuntu-22.04-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940170/job/28736162331)|
|verify-rc-source-java-macos-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10378940153/job/28736161785)|
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2320281763
@eramitmittal I have a draft PR: https://github.com/apache/arrow/pull/43880
This way you should be able to enable the logs. I checked and it works fine. Just make the changes shown in the pom.xml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2277377464
As for the memory leak, @vibhatha since we disabled the debug logging, how can it be reenabled here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2330388628
Argh, I didn't notice that this used a closed GitHub issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703598817
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/NoOpFlightSqlProducer.java:
##########
@@ -91,6 +91,15 @@ public Runnable acceptPutStatement(
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ FlightSql.CommandStatementIngest command,
+ CallContext context,
+ FlightStream flightStream,
+ StreamListener<PutResult> ackStream) {
+ throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
Review Comment:
Done for acceptPutStatementBulkIngest. However, for now, I have not touched other methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703557073
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
+ throw CallStatus.UNIMPLEMENTED.withCause(e).toRuntimeException();
+ } catch (final InterruptedException | ExecutionException e) {
Review Comment:
This is same as the code fragments in executeUpdate, executeSubstraitUpdate and execute methods:
`
} catch (final InterruptedException | ExecutionException e) {
throw CallStatus.CANCELLED.withCause(e).toRuntimeException();
} catch (final InvalidProtocolBufferException e) {
throw CallStatus.INVALID_ARGUMENT.withCause(e).toRuntimeException();
}
`
I will change all the occurances.
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
+ throw CallStatus.UNIMPLEMENTED.withCause(e).toRuntimeException();
+ } catch (final InterruptedException | ExecutionException e) {
Review Comment:
This is same as the code fragments in executeUpdate, executeSubstraitUpdate and execute methods:
`
} catch (final InterruptedException | ExecutionException e) {
throw CallStatus.CANCELLED.withCause(e).toRuntimeException();
} catch (final InvalidProtocolBufferException e) {
throw CallStatus.INVALID_ARGUMENT.withCause(e).toRuntimeException();
}
`
I will change all the occurrences.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2273078807
Sorry @eramitmittal, I will try to get to it tomorrow
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2275035249
Thanks looking into the integration failures. They passed on my local machine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328296522
failure in last run is not related to my changes:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2309505814
@vibhatha can you get it working locally? I think the debug mechanisms need to work...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2277374942
That's not the error I'm concerned about, I'm more wondering why the integration tests fail:
```
--------------
WARNING: Unknown module: org.apache.arrow.flight.core specified to --add-reads
WARNING: Unknown module: org.apache.arrow.memory.core specified to --add-opens
SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation
SLF4J(W): See https://www.slf4j.org/codes.html#noProviders for further details.
Exception in thread "main" java.lang.AssertionError: Expected:
false
but got:
null
at org.apache.arrow.flight.integration.tests.IntegrationAssertions.assertEquals(IntegrationAssertions.java:54)
at org.apache.arrow.flight.integration.tests.FlightSqlScenario.lambda$validateMetadataRetrieval$0(FlightSqlScenario.java:172)
at org.apache.arrow.flight.integration.tests.FlightSqlScenario.validate(FlightSqlScenario.java:227)
at org.apache.arrow.flight.integration.tests.FlightSqlScenario.validateMetadataRetrieval(FlightSqlScenario.java:161)
at org.apache.arrow.flight.integration.tests.FlightSqlScenario.client(FlightSqlScenario.java:76)
at org.apache.arrow.flight.integration.tests.IntegrationTestClient.run(IntegrationTestClient.java:96)
at org.apache.arrow.flight.integration.tests.IntegrationTestClient.main(IntegrationTestClient.java:66)
--------------
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2287879949
Thanks. Integration tests pass with crossbow.
The failures we see in crossbow are all what @kou highlighted.
I ran the tests close to 10 times on my local machine using the maven docker image with JDK21 and they passed each time.
Only way now is an option to turn on allocator debug for CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743277749
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
Sorry, do you mind documenting why we need the wait?
@vibhatha can we link this to a new issue? I think this is possibly the source of other test flakiness that we've been unable to track down over the years
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743454413
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
@lidavidm will create an issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325799134
nope.... same old story. Interestingly same integration test pass every single time when run as part of conda integration tests.
Now I am at loss on how to proceed.
by the way is this expected (javac 11 getting used while the label says jdk 21)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325827118
Also it seems 2 bytes leaked so possibly we could guess which buffer it was from that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703533456
##########
java/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java:
##########
@@ -363,6 +364,15 @@ public Runnable acceptPutStatement(
};
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ final CommandStatementIngest commandStatementIngest,
+ final CallContext callContext,
+ final FlightStream flightStream,
+ final StreamListener<PutResult> streamListener) {
+ throw new UnsupportedOperationException();
Review Comment:
This is in jdbc. Since jdbc does not support bulk ingestion, can you please suggest what can be done in the mock? Do you mean return 0 rows updated as PutResult?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703533456
##########
java/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java:
##########
@@ -363,6 +364,15 @@ public Runnable acceptPutStatement(
};
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ final CommandStatementIngest commandStatementIngest,
+ final CallContext callContext,
+ final FlightStream flightStream,
+ final StreamListener<PutResult> streamListener) {
+ throw new UnsupportedOperationException();
Review Comment:
After adding default implementation for FlightSqlProduce, this will automatically return unimplemented exception. I will remove explicit override in the mock
##########
java/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java:
##########
@@ -363,6 +364,15 @@ public Runnable acceptPutStatement(
};
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ final CommandStatementIngest commandStatementIngest,
+ final CallContext callContext,
+ final FlightStream flightStream,
+ final StreamListener<PutResult> streamListener) {
+ throw new UnsupportedOperationException();
Review Comment:
After adding default implementation for FlightSqlProducer, this will automatically return unimplemented exception. I will remove explicit override in the mock
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2283171566
Integration tests are passing now.
2 jobs are failing.
Java / AMD64 Ubuntu 22.04 Java JDK 17 Maven 3.9.6 (pull_request):
Failure is related to memory leak which I am unable to reproduce in my local environment. To diagnose the problem I need a CI run with allocator memory Debug turned on. Any suggestion on how to proceed?
Java JNI / AMD64 manylinux2014 Java JNI (pull_request):
These failure are strange given that I didn't touch Gandiva or Dataset projects:
<img width="881" alt="image" src="https://github.com/user-attachments/assets/c6e58580-9468-42fc-8365-37f0e5a58a8a">
Are we sure these are not some intermittent issues?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2280868361
@lidavidm the integration tests should be fixed now with new pushed commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2287418714
@github-actions crossbow submit -g java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2284926271
The failures were introduced by https://github.com/apache/arrow/pull/43503 .
The dataset failure is tracked by https://github.com/apache/arrow/issues/43506 .
The Gandiva failure isn't tracked yet. We need to use old LLVM for now. See also:
* https://github.com/apache/arrow/pull/43503#issuecomment-2264551357
* https://github.com/apache/arrow/pull/43503#issuecomment-2264965841
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325695279
@vibhatha done. Please approve CI run
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703559377
##########
java/flight/flight-integration-tests/src/main/java/org/apache/arrow/flight/integration/tests/FlightSqlScenarioProducer.java:
##########
@@ -577,35 +646,12 @@ public Runnable acceptPutPreparedStatementQuery(
@Override
public FlightInfo getFlightInfoSqlInfo(
FlightSql.CommandGetSqlInfo request, CallContext context, FlightDescriptor descriptor) {
- if (request.getInfoCount() == 2) {
Review Comment:
validateMetadataRetrieval method in FlightSqlScenario class has been enhanced to check expected results. Hence test coverage is not lost.
```
validate(
FlightSqlProducer.Schemas.GET_SQL_INFO_SCHEMA,
sqlClient.getSqlInfo(
new FlightSql.SqlInfo[] {
FlightSql.SqlInfo.FLIGHT_SQL_SERVER_NAME,
FlightSql.SqlInfo.FLIGHT_SQL_SERVER_READ_ONLY
},
options),
sqlClient,
s -> {
Map<Integer, Object> infoValues = readSqlInfoStream(s);
IntegrationAssertions.assertEquals(
Boolean.FALSE, infoValues.get(FlightSql.SqlInfo.FLIGHT_SQL_SERVER_READ_ONLY_VALUE));
IntegrationAssertions.assertEquals(
FlightSqlScenarioProducer.SERVER_NAME,
infoValues.get(FlightSql.SqlInfo.FLIGHT_SQL_SERVER_NAME_VALUE));
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2274854913
It appears this fails the actual integration tests, though
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2277360186
@lidavidm any hint on how to proceed?
Only Java 21+ runs are failing on Ubuntu complaining about memory leak in `flightSqlIngestion `test of `IntegrationTest` class.
However, I am not able to reproduce the problem.
I see a successful run when I manually run on Ubuntu with Java 21 in my local environment.
**
**
I have also tried to statically study the `FlightSqlIngestionScenario` code and I don't see how mem leak could be occurring.
Since the problem occurs only in CI, is there any possibility to run the tests in CI with DEBUG on for allocator?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2272742251
hi @lidavidm, any update please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2309508126
Yes locally when I use the flags I can see the logs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm merged PR #43551:
URL: https://github.com/apache/arrow/pull/43551
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328113864
@eramitmittal I guess then in this case we need to make sure we gracefully release all resources.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2286839843
@github-actions crossbow submit java-jars
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2288137095
Sure @lidavidm I can take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2286840972
```
Only contributors can submit requests to this bot. Please ask someone from the community for help with getting the first commit in.
The Archery job run can be found at: https://github.com/apache/arrow/actions/runs/10374976468
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2312128290
Issue created https://github.com/apache/arrow/issues/43843
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2309274786
@eramitmittal unfortunately, I couldn't get the logs working in my initial attempt after adding the flags. I was testing it in another PR but couldn't try again, but the way to do is mentioned here: https://arrow.apache.org/docs/java/memory.html#debugging-memory-leaks-allocation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743342295
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
added and pushed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743468838
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
https://github.com/apache/arrow/issues/43948
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703554444
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
Review Comment:
The private method has throws Exception declaration but the public methods using this private method don't have it. They are already catching and wrapping. e.g.
` public long executeIngest(
final ArrowStreamReader dataReader,
final ExecuteIngestOptions ingestOptions,
Transaction transaction,
final CallOption... options) {
`
Anyway, with the code to check bulk ingestion support removed, there will be no need to put a throws declaration. This will be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325750487
@vibhatha @lidavidm all tests have passed now, when I turned on the debug logs!!
May be syncing the branch with main fixed something.
Should I just revert the pom change for allocator debug and we consider this PR as ready for merge?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328024081
I see client side allocating and releasing PutResult fine.
```
allocate: 2:
org.apache.arrow.flight.integration.tests.TestBufferAllocationListener.onAllocation(TestBufferAllocationListener.java:39)
org.apache.arrow.memory.BaseAllocator.buffer(BaseAllocator.java:337)
org.apache.arrow.memory.BaseAllocator.buffer(BaseAllocator.java:297)
org.apache.arrow.flight.PutResult.fromProtocol(PutResult.java:82)
org.apache.arrow.flight.FlightClient$SetStreamObserver.onNext(FlightClient.java:488)
org.apache.arrow.flight.FlightClient$SetStreamObserver.onNext(FlightClient.java:476)
```
```
release: 2:
org.apache.arrow.flight.integration.tests.TestBufferAllocationListener.onRelease(TestBufferAllocationListener.java:43)
org.apache.arrow.memory.AllocationManager.release(AllocationManager.java:171)
org.apache.arrow.memory.BufferLedger.decrement(BufferLedger.java:153)
org.apache.arrow.memory.BufferLedger.release(BufferLedger.java:123)
org.apache.arrow.memory.BufferLedger.release(BufferLedger.java:104)
org.apache.arrow.memory.ArrowBuf.close(ArrowBuf.java:1032)
org.apache.arrow.flight.PutResult.close(PutResult.java:97)
org.apache.arrow.flight.sql.FlightSqlClient.executeIngest(FlightSqlClient.java:328)
```
Problem is at server side:
```
//FlightSqlScenarioProducer.java
private Runnable acceptPutReturnConstant(StreamListener<PutResult> ackStream, long value) {
return () -> {
final FlightSql.DoPutUpdateResult build =
FlightSql.DoPutUpdateResult.newBuilder().setRecordCount(value).build();
try (final ArrowBuf buffer = allocator.buffer(build.getSerializedSize())) {
buffer.writeBytes(build.toByteArray());
ackStream.onNext(PutResult.metadata(buffer));
ackStream.onCompleted();
}
};
}
```
I see release corresponding to ackStream.onCompleted()
```
release: 32:
org.apache.arrow.flight.integration.tests.TestBufferAllocationListener.onRelease(TestBufferAllocationListener.java:43)
...........
org.apache.arrow.flight.StreamPipe.onCompleted(StreamPipe.java:99)
org.apache.arrow.flight.integration.tests.FlightSqlScenarioProducer.lambda$acceptPutReturnConstant$0(FlightSqlScenarioProducer.java:526)
org.apache.arrow.flight.FlightService.lambda$doPutCustom$0(FlightService.java:244)
```
..but no trace of buffer getting closed due to try-with-resources. It seems as if executor thread got completed before calling the finally!!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328060720
```
/** Request that the server shut down. */
public void shutdown() {
server.shutdown();
if (grpcExecutor != null) {
grpcExecutor.shutdown();
}
}
```
ExecutorService is being shutdown without waiting for pending tasks to finish.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743325808
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
adding:
```
//Shutdown the executor while allowing existing tasks to finish.
//Without this wait, allocator.close() may get invoked earlier than an executor thread may have finished freeing up resources
//In that case, allocator.close can throw an IllegalStateException for memory leak leading to flaky tests
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328453388
> All tests passed in last run except: AMD64 manylinux2014 Java JNI
>
> Failure in that are not related to my changes:
>
> 
>
>
Yes, this is irrelevant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325826320
We could perhaps try to narrow down the source by using child allocators in certain contexts (e.g. separate server and client allocators), and/or modifying the debug code to only record a single traceback (where the buffer was originally allocated) and modifying the error to print this traceback out on failure too (the current debug code has very high overhead due to recording tracebacks nearly any time you do something with a buffer's refcount)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2309519745
Let's try to get it working in CI then so we can debug this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703430690
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java:
##########
@@ -777,6 +785,23 @@ Runnable acceptPutStatement(
FlightStream flightStream,
StreamListener<PutResult> ackStream);
+ /**
+ * Accepts uploaded data for a particular bulk ingest data stream.
+ *
+ * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+ *
+ * @param command The bulk ingestion request.
+ * @param context Per-call context.
+ * @param flightStream The data stream being uploaded.
+ * @param ackStream The result data stream.
+ * @return A runnable to process the stream.
+ */
+ Runnable acceptPutStatementBulkIngest(
Review Comment:
Can we add a `default` implementation?
##########
java/flight/flight-integration-tests/src/main/java/org/apache/arrow/flight/integration/tests/FlightSqlScenarioProducer.java:
##########
@@ -577,35 +646,12 @@ public Runnable acceptPutPreparedStatementQuery(
@Override
public FlightInfo getFlightInfoSqlInfo(
FlightSql.CommandGetSqlInfo request, CallContext context, FlightDescriptor descriptor) {
- if (request.getInfoCount() == 2) {
Review Comment:
This seems to have been unconditionally removed?
##########
java/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java:
##########
@@ -363,6 +364,15 @@ public Runnable acceptPutStatement(
};
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ final CommandStatementIngest commandStatementIngest,
+ final CallContext callContext,
+ final FlightStream flightStream,
+ final StreamListener<PutResult> streamListener) {
+ throw new UnsupportedOperationException();
Review Comment:
We should be using `streamListener` instead of throwing.
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
Review Comment:
Why does this `throws Exception` when that's going to force every other overload to catch and wrap it, when we could catch and wrap it here?
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
Review Comment:
I think it's better to just execute it and let the server decide what is or isn't supported
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/NoOpFlightSqlProducer.java:
##########
@@ -91,6 +91,15 @@ public Runnable acceptPutStatement(
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ FlightSql.CommandStatementIngest command,
+ CallContext context,
+ FlightStream flightStream,
+ StreamListener<PutResult> ackStream) {
+ throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
Review Comment:
Use the `StreamListener` instead of throwing.
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -1003,6 +1185,82 @@ public void close() throws Exception {
AutoCloseables.close(client);
}
+ /** Class to encapsulateFlight SQL bulk ingest request options. * */
Review Comment:
```suggestion
/** Class to encapsulate Flight SQL bulk ingest request options. * */
```
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
+ throw CallStatus.UNIMPLEMENTED.withCause(e).toRuntimeException();
+ } catch (final InterruptedException | ExecutionException e) {
Review Comment:
It might make more sense to unwrap the ExecutionException instead?
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
Review Comment:
Is this actually thrown?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703535937
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java:
##########
@@ -777,6 +785,23 @@ Runnable acceptPutStatement(
FlightStream flightStream,
StreamListener<PutResult> ackStream);
+ /**
+ * Accepts uploaded data for a particular bulk ingest data stream.
+ *
+ * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+ *
+ * @param command The bulk ingestion request.
+ * @param context Per-call context.
+ * @param flightStream The data stream being uploaded.
+ * @param ackStream The result data stream.
+ * @return A runnable to process the stream.
+ */
+ Runnable acceptPutStatementBulkIngest(
Review Comment:
Ok I will add : ackStream.onError(CallStatus.UNIMPLEMENTED.toRuntimeException()); as default implementation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703557562
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -1003,6 +1185,82 @@ public void close() throws Exception {
AutoCloseables.close(client);
}
+ /** Class to encapsulateFlight SQL bulk ingest request options. * */
Review Comment:
thanks for spotting it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703559377
##########
java/flight/flight-integration-tests/src/main/java/org/apache/arrow/flight/integration/tests/FlightSqlScenarioProducer.java:
##########
@@ -577,35 +646,12 @@ public Runnable acceptPutPreparedStatementQuery(
@Override
public FlightInfo getFlightInfoSqlInfo(
FlightSql.CommandGetSqlInfo request, CallContext context, FlightDescriptor descriptor) {
- if (request.getInfoCount() == 2) {
Review Comment:
validateMetadataRetrieval method in FlightSqlScenario class has been enhanced to check expected results. Hence test coverage is not lost.
`
validate(
FlightSqlProducer.Schemas.GET_SQL_INFO_SCHEMA,
sqlClient.getSqlInfo(
new FlightSql.SqlInfo[] {
FlightSql.SqlInfo.FLIGHT_SQL_SERVER_NAME,
FlightSql.SqlInfo.FLIGHT_SQL_SERVER_READ_ONLY
},
options),
sqlClient,
s -> {
Map<Integer, Object> infoValues = readSqlInfoStream(s);
IntegrationAssertions.assertEquals(
Boolean.FALSE, infoValues.get(FlightSql.SqlInfo.FLIGHT_SQL_SERVER_READ_ONLY_VALUE));
IntegrationAssertions.assertEquals(
FlightSqlScenarioProducer.SERVER_NAME,
infoValues.get(FlightSql.SqlInfo.FLIGHT_SQL_SERVER_NAME_VALUE));
});
`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703597729
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java:
##########
@@ -777,6 +785,23 @@ Runnable acceptPutStatement(
FlightStream flightStream,
StreamListener<PutResult> ackStream);
+ /**
+ * Accepts uploaded data for a particular bulk ingest data stream.
+ *
+ * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+ *
+ * @param command The bulk ingestion request.
+ * @param context Per-call context.
+ * @param flightStream The data stream being uploaded.
+ * @param ackStream The result data stream.
+ * @return A runnable to process the stream.
+ */
+ Runnable acceptPutStatementBulkIngest(
Review Comment:
Done. Included new commit with this change
##########
java/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java:
##########
@@ -363,6 +364,15 @@ public Runnable acceptPutStatement(
};
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ final CommandStatementIngest commandStatementIngest,
+ final CallContext callContext,
+ final FlightStream flightStream,
+ final StreamListener<PutResult> streamListener) {
+ throw new UnsupportedOperationException();
Review Comment:
Done. Included new commit with this change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2309558389
> Let's try to get it working in CI then so we can debug this.
I will create an issue and work on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328186518
Pushed a wait for ExecutorService to terminate in IntegrationTest.testScenario method. @lidavidm please approve the CI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2326355286
@lidavidm I have added an allocationListener to the tests to keep track of allocation and release. In case of failure due to IllegalStateException I am then printing the allocation/release trail. Hopefully this should be able to provide information on where the mem leak occurs. Can you please approve the CI?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2282943648
> As for the memory leak, @vibhatha since we disabled the debug logging, how can it be reenabled here?
I need to take a look @lidavidm, I haven't enabled it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2288494264
I am looking into this, I will post the steps once it is done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2286252859
@kou thanks.
So, I just have to worry about mem leak failure.
```java.lang.IllegalStateException:
Memory was leaked by query. Memory leaked: (2)
Allocator(ROOT) 0/2/32804/9223372036854775807 (res/actual/peak/limit)
at org.apache.arrow.memory.BaseAllocator.close(BaseAllocator.java:503)
at org.apache.arrow.memory.RootAllocator.close(RootAllocator.java:27)
at org.apache.arrow.flight.integration.tests.IntegrationTest.testScenario(IntegrationTest.java:116)
at org.apache.arrow.flight.integration.tests.IntegrationTest.flightSqlIngestion(IntegrationTest.java:85)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)```
Strangely sometimes it fails with the above error for JDK 17 and sometimes for JDK 21. It seems some kind of race condition. Let me try crossbow and see if it fails. Also, locally I will try to have multiple runs using same docker images as is being used here and try to reproduce the failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2296360973
hi @vibhatha any leads?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743308457
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
Document as in add a comment in the code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743342295
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328624931
So nice to see the integration tests passing consistently now :). Hope to see this merged soon. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2327716680
Done. It seems it's either the server or the client's handling of putResult that's the issue? (Maybe sometimes we aren't draining the stream on the client properly?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703538791
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/NoOpFlightSqlProducer.java:
##########
@@ -91,6 +91,15 @@ public Runnable acceptPutStatement(
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ FlightSql.CommandStatementIngest command,
+ CallContext context,
+ FlightStream flightStream,
+ StreamListener<PutResult> ackStream) {
+ throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
Review Comment:
Will do the change. However this class lacks consistency and a lot of methods are throwing.
for e.g:
`public Runnable acceptPutStatement(
FlightSql.CommandStatementUpdate command,
CallContext context,
FlightStream flightStream,
StreamListener<PutResult> ackStream) {
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}`
or
` @Override
public FlightInfo getFlightInfoStatement(
FlightSql.CommandStatementQuery command, CallContext context, FlightDescriptor descriptor) {
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}`
should I convert all to listener.OnError?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703557073
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
+ throw CallStatus.UNIMPLEMENTED.withCause(e).toRuntimeException();
+ } catch (final InterruptedException | ExecutionException e) {
Review Comment:
This is same as the code fragments in executeUpdate, executeSubstraitUpdate and execute methods:
```
} catch (final InterruptedException | ExecutionException e) {
throw CallStatus.CANCELLED.withCause(e).toRuntimeException();
} catch (final InvalidProtocolBufferException e) {
throw CallStatus.INVALID_ARGUMENT.withCause(e).toRuntimeException();
}
```
I will change all the occurrences.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703538791
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/NoOpFlightSqlProducer.java:
##########
@@ -91,6 +91,15 @@ public Runnable acceptPutStatement(
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ FlightSql.CommandStatementIngest command,
+ CallContext context,
+ FlightStream flightStream,
+ StreamListener<PutResult> ackStream) {
+ throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
Review Comment:
Will do the change. However this class lacks consistency and a lot of methods are throwing.
for e.g:
```
public Runnable acceptPutStatement(
FlightSql.CommandStatementUpdate command,
CallContext context,
FlightStream flightStream,
StreamListener<PutResult> ackStream) {
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}
```
or
```
@Override
public FlightInfo getFlightInfoStatement(
FlightSql.CommandStatementQuery command, CallContext context, FlightDescriptor descriptor) {
throw CallStatus.UNIMPLEMENTED.withDescription("Not implemented.").toRuntimeException();
}
```
should I convert all to listener.OnError?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703599379
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
Review Comment:
Done. Included new commit with this change
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
Review Comment:
Done. Included new commit with this change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325750956
Sure, let's try it and see.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1743325808
##########
java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java:
##########
@@ -114,6 +124,8 @@ void testScenario(String scenarioName) throws Exception {
scenario.client(allocator, location, client);
}
}
+ exec.shutdown();
+ final boolean unused = exec.awaitTermination(3, TimeUnit.SECONDS);
Review Comment:
adding:
```
// Shutdown the executor while allowing existing tasks to finish.
// Without this wait, allocator.close() may get invoked earlier than an executor thread may
// have finished freeing up resources
// In that case, allocator.close() can throw an IllegalStateException for memory leak, leading
// to flaky tests
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "vibhatha (via GitHub)" <gi...@apache.org>.
vibhatha commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2325698159
> @vibhatha done. Please approve CI run
@lidavidm please approve the CI run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2328139610
@vibhatha in general FlightServer.close() and shutdown methods need a review to account for pending tasks in ExecutorService as well. Other way obviously is that I can introduce a short delay in the IntegrationTest.testScenario method to allow ExecutorService threads to finish. But that will not solve the problem getting highlighted in FlightServer. I think graceful shutdown of FlightServer can be handled as a separate issue and for now I can just make the tests pass. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "conbench-apache-arrow[bot] (via GitHub)" <gi...@apache.org>.
conbench-apache-arrow[bot] commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2330880766
After merging your PR, Conbench analyzed the 4 benchmarking runs that have been run so far on merge-commit 5ca12bd7c109cdc362f929ada12d4de79148a1c7.
There were no benchmark performance regressions. 🎉
The [full Conbench report](https://github.com/apache/arrow/runs/29713899134) has more details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703542296
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
Review Comment:
ok. Will do this change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703553928
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
Review Comment:
With the code to check bulk ingestion support removed, this code will get removed as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703554444
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
Review Comment:
The private method has throws Exception declaration but the public methods using this private method don't have it. They are already catching and wrapping. e.g.
```
public long executeIngest(
final ArrowStreamReader dataReader,
final ExecuteIngestOptions ingestOptions,
Transaction transaction,
final CallOption... options) {
```
Anyway, with the code to check bulk ingestion support removed, there will be no need to put a throws declaration. This will be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703599571
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
Review Comment:
Done. Included new commit with this change
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -206,6 +216,178 @@ public SchemaResult getExecuteSubstraitSchema(
return getExecuteSubstraitSchema(substraitPlan, /*transaction*/ null, options);
}
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(data, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ final CallOption... options) {
+ return executeIngest(dataReader, ingestOptions, /*transaction*/ null, options);
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param data data to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+ try {
+ return executeIngest(
+ data, ingestOptions, transaction, FlightClient.ClientStreamListener::putNext, options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ /**
+ * Execute a bulk ingest on the server.
+ *
+ * @param dataReader data stream to be ingested
+ * @param ingestOptions options for the ingest request.
+ * @param transaction The transaction that this ingest request is part of.
+ * @param options RPC-layer hints for this call.
+ * @return the number of rows affected.
+ */
+ public long executeIngest(
+ final ArrowStreamReader dataReader,
+ final ExecuteIngestOptions ingestOptions,
+ Transaction transaction,
+ final CallOption... options) {
+
+ try {
+ return executeIngest(
+ dataReader.getVectorSchemaRoot(),
+ ingestOptions,
+ transaction,
+ listener -> {
+ while (true) {
+ try {
+ if (!dataReader.loadNextBatch()) {
+ break;
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ listener.putNext();
+ }
+ },
+ options);
+ } catch (FlightRuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw CallStatus.UNKNOWN.withCause(e).toRuntimeException();
+ }
+ }
+
+ private long executeIngest(
+ final VectorSchemaRoot data,
+ final ExecuteIngestOptions ingestOptions,
+ final Transaction transaction,
+ final Consumer<FlightClient.ClientStreamListener> dataPutter,
+ final CallOption... options)
+ throws Exception {
+ try (FlightStream sqlInfoStream =
+ getStream(
+ getSqlInfo(
+ SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION,
+ SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED)
+ .getEndpoints()
+ .get(0)
+ .getTicket())) {
+ boolean bulkIngestionSupported = false;
+ boolean transactionSupported = false;
+
+ while (sqlInfoStream.next()) {
+ VectorSchemaRoot sqlInfoRoot = sqlInfoStream.getRoot();
+
+ UInt4Vector infoName = (UInt4Vector) sqlInfoRoot.getVector(0);
+ DenseUnionVector value = (DenseUnionVector) sqlInfoRoot.getVector(1);
+
+ for (int i = 0; i < sqlInfoRoot.getRowCount(); i++) {
+ final int code = infoName.get(i);
+ if (code == SqlInfo.FLIGHT_SQL_SERVER_BULK_INGESTION_VALUE) {
+ bulkIngestionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ } else if (code == SqlInfo.FLIGHT_SQL_SERVER_INGEST_TRANSACTIONS_SUPPORTED_VALUE) {
+ transactionSupported =
+ Boolean.TRUE.equals(
+ value.getBitVector(value.getTypeId(i)).getObject(value.getOffset(i)));
+ }
+ }
+ }
+
+ if (!bulkIngestionSupported) {
+ throw new UnsupportedOperationException("Executing bulk ingestion is not supported.");
+ }
+
+ if (!isNull(transaction) && !transactionSupported) {
+ throw new UnsupportedOperationException(
+ "Executing bulk ingestion with-in a transaction is not supported.");
+ }
+
+ final CommandStatementIngest.Builder builder = CommandStatementIngest.newBuilder();
+ if (transaction != null) {
+ builder.setTransactionId(ByteString.copyFrom(transaction.getTransactionId()));
+ }
+ ingestOptions.updateCommandBuilder(builder);
+
+ final FlightDescriptor descriptor =
+ FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+ try (final SyncPutListener putListener = new SyncPutListener()) {
+
+ final FlightClient.ClientStreamListener listener =
+ client.startPut(descriptor, data, putListener, options);
+ dataPutter.accept(listener);
+ listener.completed();
+ listener.getResult();
+
+ try (final PutResult result = putListener.read()) {
+ final DoPutUpdateResult doPutUpdateResult =
+ DoPutUpdateResult.parseFrom(result.getApplicationMetadata().nioBuffer());
+ return doPutUpdateResult.getRecordCount();
+ }
+ }
+ } catch (final UnsupportedOperationException e) {
+ throw CallStatus.UNIMPLEMENTED.withCause(e).toRuntimeException();
+ } catch (final InterruptedException | ExecutionException e) {
Review Comment:
Done. Included new commit with this change
##########
java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java:
##########
@@ -1003,6 +1185,82 @@ public void close() throws Exception {
AutoCloseables.close(client);
}
+ /** Class to encapsulateFlight SQL bulk ingest request options. * */
Review Comment:
Done. Included new commit with this change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #43551:
URL: https://github.com/apache/arrow/pull/43551#issuecomment-2288064870
@vibhatha can you show how to do that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38255: [Java] Implement Flight SQL Bulk Ingestion [arrow]
Posted by "eramitmittal (via GitHub)" <gi...@apache.org>.
eramitmittal commented on code in PR #43551:
URL: https://github.com/apache/arrow/pull/43551#discussion_r1703533456
##########
java/flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/utils/MockFlightSqlProducer.java:
##########
@@ -363,6 +364,15 @@ public Runnable acceptPutStatement(
};
}
+ @Override
+ public Runnable acceptPutStatementBulkIngest(
+ final CommandStatementIngest commandStatementIngest,
+ final CallContext callContext,
+ final FlightStream flightStream,
+ final StreamListener<PutResult> streamListener) {
+ throw new UnsupportedOperationException();
Review Comment:
This is in jdbc. Since jdbc does not support bulk ingestion, can you please suggest what can be done in the mock? Do you mean return 0 rows updated as PutResult or returning streamListener.onError(CallStatus.UNIMPLEMENTED.toRuntimeException());?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org