You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2020/12/18 19:54:06 UTC
[GitHub] [arrow] lidavidm opened a new pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
lidavidm opened a new pull request #8963:
URL: https://github.com/apache/arrow/pull/8963
Fixes compatibility with standard Protobuf implementations, which may omit serialization of an empty field. In the case of an empty record batch, this would cause the Java client to fail a precondition check since there would be no body buffer instead of a single empty body buffer.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] alamb commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-752936953
Requesting review from @praveenbingo and @liyafan82 as they seem to have merged the last few Java PRs into the arrow repo.
The background here is that we are working on flight integration tests in Rust (e.g. https://github.com/apache/arrow/pull/9049) and we found an edge case for flight integration.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 closed pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 closed pull request #8963:
URL: https://github.com/apache/arrow/pull/8963
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] alamb edited a comment on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
alamb edited a comment on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-752936953
Requesting review from @praveenbingo and @liyafan82 as they seem to have merged the last few Java PRs into the arrow repo.
The background here is that we are working on flight integration tests in Rust (e.g. https://github.com/apache/arrow/pull/9049) and we found an edge case in the flight protobuf integration.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-748291786
https://issues.apache.org/jira/browse/ARROW-10962
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551867296
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,71 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufRecordBatchCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
+ }
+
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufSchemaCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Flight.FlightDescriptor descriptor = FlightDescriptor.command(new byte[0]).toProtocol();
+ try (final ArrowMessage message = new ArrowMessage(descriptor, schema, new IpcOption())) {
Review comment:
Do we need to check the message body 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
lidavidm commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551954971
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,71 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufRecordBatchCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
+ }
+
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufSchemaCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Flight.FlightDescriptor descriptor = FlightDescriptor.command(new byte[0]).toProtocol();
+ try (final ArrowMessage message = new ArrowMessage(descriptor, schema, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.SCHEMA, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage)
+ .toBuilder()
+ .setDataBody(ByteString.EMPTY)
+ .build();
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asSchema();
Review comment:
The schema itself has no body. I've added a test to check the parsed message.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551293114
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,39 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10939: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufCompat() throws Exception {
Review comment:
Compat -> Compact?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] alamb commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-756711085
🎉 thanks @liyafan82 -- I didn't do anything, it was all @lidavidm . Thanks @lidavidm for the efforts!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] alamb commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-756711085
🎉 thanks @liyafan82 -- I didn't do anything, it was all @lidavidm . Thanks @lidavidm for the efforts!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-753717290
Sorry for my late response. Will take a look late today.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551866948
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,71 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufRecordBatchCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
+ }
+
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufSchemaCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Flight.FlightDescriptor descriptor = FlightDescriptor.command(new byte[0]).toProtocol();
+ try (final ArrowMessage message = new ArrowMessage(descriptor, schema, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.SCHEMA, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
Review comment:
We can extract a common method for the logic of writing the message to a ByteArrayOutputStream?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
lidavidm commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551326743
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,39 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10939: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufCompat() throws Exception {
Review comment:
Expanded to 'compatibility'
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,39 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10939: accept FlightData messages generated by Protobuf (which can omit empty fields). */
Review comment:
Good catch, fixed.
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,39 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10939: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufCompat() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
Review comment:
Yup, I added a test for this 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] lidavidm commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
lidavidm commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551959223
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,71 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufRecordBatchCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
+ }
+
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufSchemaCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Flight.FlightDescriptor descriptor = FlightDescriptor.command(new byte[0]).toProtocol();
+ try (final ArrowMessage message = new ArrowMessage(descriptor, schema, new IpcOption())) {
Review comment:
I've added assertions here.
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,71 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufRecordBatchCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
+ }
+
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufSchemaCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Flight.FlightDescriptor descriptor = FlightDescriptor.command(new byte[0]).toProtocol();
+ try (final ArrowMessage message = new ArrowMessage(descriptor, schema, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.SCHEMA, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
Review comment:
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551295036
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,39 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10939: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufCompat() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
Review comment:
nit: do you think we also need a test case for which a schema has an empty body?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551292760
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,39 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10939: accept FlightData messages generated by Protobuf (which can omit empty fields). */
Review comment:
10939 -> 10962?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] alamb commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
alamb commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-756239945
https://github.com/apache/arrow/pull/9049 is still blocked on getting this PR merged. Any chance you can do so @liyafan82 ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on a change in pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on a change in pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#discussion_r551866156
##########
File path: java/flight/flight-core/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
##########
@@ -317,6 +325,71 @@ private void test(BiConsumer<FlightClient, BufferAllocator> consumer) throws Exc
}
}
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufRecordBatchCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
+ final VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator)) {
+ final VectorUnloader unloader = new VectorUnloader(root);
+ root.setRowCount(0);
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final ArrowMessage message = new ArrowMessage(unloader.getRecordBatch(), null, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.RECORD_BATCH, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage);
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ final ArrowRecordBatch rb =
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asRecordBatch();
+ Assert.assertEquals(rb.computeBodyLength(), 0);
+ }
+ }
+
+ /** ARROW-10962: accept FlightData messages generated by Protobuf (which can omit empty fields). */
+ @Test
+ public void testProtobufSchemaCompatibility() throws Exception {
+ final Schema schema = new Schema(Collections.singletonList(Field.nullable("foo", new ArrowType.Int(32, true))));
+ try (final BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE)) {
+ final MethodDescriptor.Marshaller<ArrowMessage> marshaller = ArrowMessage.createMarshaller(allocator);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ Flight.FlightDescriptor descriptor = FlightDescriptor.command(new byte[0]).toProtocol();
+ try (final ArrowMessage message = new ArrowMessage(descriptor, schema, new IpcOption())) {
+ Assert.assertEquals(ArrowMessage.HeaderType.SCHEMA, message.getMessageType());
+ try (final InputStream serialized = marshaller.stream(message)) {
+ final byte[] buf = new byte[1024];
+ while (true) {
+ int read = serialized.read(buf);
+ if (read < 0) {
+ break;
+ }
+ baos.write(buf, 0, read);
+ }
+ }
+ }
+ final byte[] serializedMessage = baos.toByteArray();
+ final Flight.FlightData protobufData = Flight.FlightData.parseFrom(serializedMessage)
+ .toBuilder()
+ .setDataBody(ByteString.EMPTY)
+ .build();
+ Assert.assertEquals(0, protobufData.getDataBody().size());
+ // Should not throw
+ marshaller.parse(new ByteArrayInputStream(protobufData.toByteArray())).asSchema();
Review comment:
Maybe we need to assign the schema to an object and verify that its body is null?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] liyafan82 commented on pull request #8963: ARROW-10962: [FlightRPC][Java] fill in empty body buffer if needed
Posted by GitBox <gi...@apache.org>.
liyafan82 commented on pull request #8963:
URL: https://github.com/apache/arrow/pull/8963#issuecomment-756495843
Merging. Thanks for your effort. @lidavidm @alamb
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org