You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ap...@apache.org on 2019/07/25 16:01:38 UTC
[arrow] branch master updated: ARROW-5979: [FlightRPC] Expose
opaque (de)serialization of protocol types
This is an automated email from the ASF dual-hosted git repository.
apitrou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new b7a5bcc ARROW-5979: [FlightRPC] Expose opaque (de)serialization of protocol types
b7a5bcc is described below
commit b7a5bcc581634cd3b1e649d8a02749db2d5f8d7f
Author: David Li <li...@gmail.com>
AuthorDate: Thu Jul 25 18:01:20 2019 +0200
ARROW-5979: [FlightRPC] Expose opaque (de)serialization of protocol types
Motivation: make it easier for non-Flight services to consume/produce Flight types. For instance, a REST service may want to produce FlightInfo or Ticket objects for a Flight client. This gives us an (opaque) serialization of a few protocol types that can be deserialized and then used as normal.
I've noticed some of the Java APIs accidentally expose conversions to/from Protobuf types; if it's worth it, I can go back and make those private.
Personal Travis: https://travis-ci.com/lihalite/arrow/builds/119836432
Closes #4914 from lihalite/arrow-5979 and squashes the following commits:
14cc13bd5 <David Li> Expose Flight protocol types in Python
c17160af8 <David Li> Expose Flight protocol types in C++
5662c4a20 <David Li> Expose Flight protocol types in Java
Authored-by: David Li <li...@gmail.com>
Signed-off-by: Antoine Pitrou <an...@python.org>
---
cpp/src/arrow/flight/flight-test.cc | 42 ++++++++++
cpp/src/arrow/flight/types.cc | 66 +++++++++++++++
cpp/src/arrow/flight/types.h | 63 ++++++++++++++
.../main/java/org/apache/arrow/flight/Action.java | 2 +-
.../org/apache/arrow/flight/FlightDescriptor.java | 27 +++++-
.../org/apache/arrow/flight/FlightEndpoint.java | 30 ++++++-
.../java/org/apache/arrow/flight/FlightInfo.java | 58 +++++++++++++
.../java/org/apache/arrow/flight/Location.java | 28 ++++++-
.../main/java/org/apache/arrow/flight/Result.java | 2 +-
.../main/java/org/apache/arrow/flight/Ticket.java | 24 ++++++
.../apache/arrow/flight/TestBasicOperation.java | 47 +++++++++++
python/pyarrow/_flight.pyx | 96 +++++++++++++++++++++-
python/pyarrow/includes/libarrow_flight.pxd | 15 ++++
python/pyarrow/tests/test_flight.py | 32 ++++++++
14 files changed, 526 insertions(+), 6 deletions(-)
diff --git a/cpp/src/arrow/flight/flight-test.cc b/cpp/src/arrow/flight/flight-test.cc
index 63c7546..901f626 100644
--- a/cpp/src/arrow/flight/flight-test.cc
+++ b/cpp/src/arrow/flight/flight-test.cc
@@ -179,6 +179,48 @@ TEST(TestFlight, ConnectUri) {
ASSERT_OK(FlightClient::Connect(location2, &client));
}
+TEST(TestFlight, RoundTripTypes) {
+ Ticket ticket{"foo"};
+ std::string ticket_serialized;
+ Ticket ticket_deserialized;
+ ASSERT_OK(ticket.SerializeToString(&ticket_serialized));
+ ASSERT_OK(Ticket::Deserialize(ticket_serialized, &ticket_deserialized));
+ ASSERT_EQ(ticket.ticket, ticket_deserialized.ticket);
+
+ FlightDescriptor desc = FlightDescriptor::Command("select * from foo;");
+ std::string desc_serialized;
+ FlightDescriptor desc_deserialized;
+ ASSERT_OK(desc.SerializeToString(&desc_serialized));
+ ASSERT_OK(FlightDescriptor::Deserialize(desc_serialized, &desc_deserialized));
+ ASSERT_TRUE(desc.Equals(desc_deserialized));
+
+ desc = FlightDescriptor::Path({"a", "b", "test.arrow"});
+ ASSERT_OK(desc.SerializeToString(&desc_serialized));
+ ASSERT_OK(FlightDescriptor::Deserialize(desc_serialized, &desc_deserialized));
+ ASSERT_TRUE(desc.Equals(desc_deserialized));
+
+ FlightInfo::Data data;
+ std::shared_ptr<Schema> schema =
+ arrow::schema({field("a", int64()), field("b", int64()), field("c", int64()),
+ field("d", int64())});
+ Location location1, location2, location3;
+ ASSERT_OK(Location::ForGrpcTcp("localhost", 10010, &location1));
+ ASSERT_OK(Location::ForGrpcTls("localhost", 10010, &location2));
+ ASSERT_OK(Location::ForGrpcUnix("/tmp/test.sock", &location3));
+ std::vector<FlightEndpoint> endpoints{FlightEndpoint{ticket, {location1, location2}},
+ FlightEndpoint{ticket, {location3}}};
+ ASSERT_OK(MakeFlightInfo(*schema, desc, endpoints, -1, -1, &data));
+ std::unique_ptr<FlightInfo> info = std::unique_ptr<FlightInfo>(new FlightInfo(data));
+ std::string info_serialized;
+ std::unique_ptr<FlightInfo> info_deserialized;
+ ASSERT_OK(info->SerializeToString(&info_serialized));
+ ASSERT_OK(FlightInfo::Deserialize(info_serialized, &info_deserialized));
+ ASSERT_TRUE(info->descriptor().Equals(info_deserialized->descriptor()));
+ ASSERT_EQ(info->endpoints(), info_deserialized->endpoints());
+ ASSERT_EQ(info->total_records(), info_deserialized->total_records());
+ ASSERT_EQ(info->total_bytes(), info_deserialized->total_bytes());
+}
+
// ----------------------------------------------------------------------
// Client tests
diff --git a/cpp/src/arrow/flight/types.cc b/cpp/src/arrow/flight/types.cc
index 86aa223..89ebd82 100644
--- a/cpp/src/arrow/flight/types.cc
+++ b/cpp/src/arrow/flight/types.cc
@@ -21,6 +21,7 @@
#include <sstream>
#include <utility>
+#include "arrow/flight/serialization-internal.h"
#include "arrow/io/memory.h"
#include "arrow/ipc/dictionary.h"
#include "arrow/ipc/reader.h"
@@ -77,6 +78,45 @@ std::string FlightDescriptor::ToString() const {
return ss.str();
}
+Status FlightDescriptor::SerializeToString(std::string* out) const {
+ pb::FlightDescriptor pb_descriptor;
+ RETURN_NOT_OK(internal::ToProto(*this, &pb_descriptor));
+
+ if (!pb_descriptor.SerializeToString(out)) {
+ return Status::IOError("Serialized descriptor exceeded 2 GiB limit");
+ }
+ return Status::OK();
+}
+
+Status FlightDescriptor::Deserialize(const std::string& serialized,
+ FlightDescriptor* out) {
+ pb::FlightDescriptor pb_descriptor;
+ if (!pb_descriptor.ParseFromString(serialized)) {
+ return Status::Invalid("Not a valid descriptor");
+ }
+ return internal::FromProto(pb_descriptor, out);
+}
+
+bool Ticket::Equals(const Ticket& other) const { return ticket == other.ticket; }
+
+Status Ticket::SerializeToString(std::string* out) const {
+ pb::Ticket pb_ticket;
+ internal::ToProto(*this, &pb_ticket);
+
+ if (!pb_ticket.SerializeToString(out)) {
+ return Status::IOError("Serialized ticket exceeded 2 GiB limit");
+ }
+ return Status::OK();
+}
+
+Status Ticket::Deserialize(const std::string& serialized, Ticket* out) {
+ pb::Ticket pb_ticket;
+ if (!pb_ticket.ParseFromString(serialized)) {
+ return Status::Invalid("Not a valid ticket");
+ }
+ return internal::FromProto(pb_ticket, out);
+}
+
Status FlightInfo::GetSchema(ipc::DictionaryMemo* dictionary_memo,
std::shared_ptr<Schema>* out) const {
if (reconstructed_schema_) {
@@ -90,6 +130,28 @@ Status FlightInfo::GetSchema(ipc::DictionaryMemo* dictionary_memo,
return Status::OK();
}
+Status FlightInfo::SerializeToString(std::string* out) const {
+ pb::FlightInfo pb_info;
+ RETURN_NOT_OK(internal::ToProto(*this, &pb_info));
+
+ if (!pb_info.SerializeToString(out)) {
+ return Status::IOError("Serialized FlightInfo exceeded 2 GiB limit");
+ }
+ return Status::OK();
+}
+
+Status FlightInfo::Deserialize(const std::string& serialized,
+ std::unique_ptr<FlightInfo>* out) {
+ pb::FlightInfo pb_info;
+ if (!pb_info.ParseFromString(serialized)) {
+ return Status::Invalid("Not a valid FlightInfo");
+ }
+ FlightInfo::Data data;
+ RETURN_NOT_OK(internal::FromProto(pb_info, &data));
+ out->reset(new FlightInfo(data));
+ return Status::OK();
+}
+
Location::Location() { uri_ = std::make_shared<arrow::internal::Uri>(); }
Status Location::Parse(const std::string& uri_string, Location* location) {
@@ -128,6 +190,10 @@ bool Location::Equals(const Location& other) const {
return ToString() == other.ToString();
}
+bool FlightEndpoint::Equals(const FlightEndpoint& other) const {
+ return ticket == other.ticket && locations == other.locations;
+}
+
Status MetadataRecordBatchReader::ReadAll(
std::vector<std::shared_ptr<RecordBatch>>* batches) {
FlightStreamChunk chunk;
diff --git a/cpp/src/arrow/flight/types.h b/cpp/src/arrow/flight/types.h
index b4c4c6c..a80b697 100644
--- a/cpp/src/arrow/flight/types.h
+++ b/cpp/src/arrow/flight/types.h
@@ -113,8 +113,21 @@ struct ARROW_FLIGHT_EXPORT FlightDescriptor {
bool Equals(const FlightDescriptor& other) const;
+ /// \brief Get a human-readable form of this descriptor.
std::string ToString() const;
+ /// \brief Get the wire-format representation of this type.
+ ///
+ /// Useful when interoperating with non-Flight systems (e.g. REST
+ /// services) that may want to return Flight types.
+ Status SerializeToString(std::string* out) const;
+
+ /// \brief Parse the wire-format representation of this type.
+ ///
+ /// Useful when interoperating with non-Flight systems (e.g. REST
+ /// services) that may want to return Flight types.
+ static Status Deserialize(const std::string& serialized, FlightDescriptor* out);
+
// Convenience factory functions
static FlightDescriptor Command(const std::string& c) {
@@ -124,12 +137,40 @@ struct ARROW_FLIGHT_EXPORT FlightDescriptor {
static FlightDescriptor Path(const std::vector<std::string>& p) {
return FlightDescriptor{PATH, "", p};
}
+
+ friend bool operator==(const FlightDescriptor& left, const FlightDescriptor& right) {
+ return left.Equals(right);
+ }
+ friend bool operator!=(const FlightDescriptor& left, const FlightDescriptor& right) {
+ return !(left == right);
+ }
};
/// \brief Data structure providing an opaque identifier or credential to use
/// when requesting a data stream with the DoGet RPC
struct ARROW_FLIGHT_EXPORT Ticket {
std::string ticket;
+
+ bool Equals(const Ticket& other) const;
+
+ friend bool operator==(const Ticket& left, const Ticket& right) {
+ return left.Equals(right);
+ }
+ friend bool operator!=(const Ticket& left, const Ticket& right) {
+ return !(left == right);
+ }
+
+ /// \brief Get the wire-format representation of this type.
+ ///
+ /// Useful when interoperating with non-Flight systems (e.g. REST
+ /// services) that may want to return Flight types.
+ Status SerializeToString(std::string* out) const;
+
+ /// \brief Parse the wire-format representation of this type.
+ ///
+ /// Useful when interoperating with non-Flight systems (e.g. REST
+ /// services) that may want to return Flight types.
+ static Status Deserialize(const std::string& serialized, Ticket* out);
};
class FlightClient;
@@ -204,6 +245,15 @@ struct ARROW_FLIGHT_EXPORT FlightEndpoint {
/// ticket can only be redeemed on the current service where the ticket was
/// generated
std::vector<Location> locations;
+
+ bool Equals(const FlightEndpoint& other) const;
+
+ friend bool operator==(const FlightEndpoint& left, const FlightEndpoint& right) {
+ return left.Equals(right);
+ }
+ friend bool operator!=(const FlightEndpoint& left, const FlightEndpoint& right) {
+ return !(left == right);
+ }
};
/// \brief Staging data structure for messages about to be put on the wire
@@ -255,6 +305,19 @@ class ARROW_FLIGHT_EXPORT FlightInfo {
/// The total number of bytes in the dataset. If unknown, set to -1
int64_t total_bytes() const { return data_.total_bytes; }
+ /// \brief Get the wire-format representation of this type.
+ ///
+ /// Useful when interoperating with non-Flight systems (e.g. REST
+ /// services) that may want to return Flight types.
+ Status SerializeToString(std::string* out) const;
+
+ /// \brief Parse the wire-format representation of this type.
+ ///
+ /// Useful when interoperating with non-Flight systems (e.g. REST
+ /// services) that may want to return Flight types.
+ static Status Deserialize(const std::string& serialized,
+ std::unique_ptr<FlightInfo>* out);
+
private:
Data data_;
mutable std::shared_ptr<Schema> schema_;
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/Action.java b/java/flight/src/main/java/org/apache/arrow/flight/Action.java
index fe7a9ce..524ffca 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/Action.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/Action.java
@@ -40,7 +40,7 @@ public class Action {
this.body = body == null ? new byte[0] : body;
}
- public Action(Flight.Action action) {
+ Action(Flight.Action action) {
this(action.getType(), action.getBody().toByteArray());
}
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/FlightDescriptor.java b/java/flight/src/main/java/org/apache/arrow/flight/FlightDescriptor.java
index ac2fff3..3f9e6bd 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/FlightDescriptor.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/FlightDescriptor.java
@@ -17,6 +17,9 @@
package org.apache.arrow.flight;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.List;
import org.apache.arrow.flight.impl.Flight;
@@ -92,6 +95,28 @@ public class FlightDescriptor {
return b.setType(DescriptorType.PATH).addAllPath(path).build();
}
+ /**
+ * Get the serialized form of this protocol message.
+ *
+ * <p>Intended to help interoperability by allowing non-Flight services to still return Flight types.
+ */
+ public ByteBuffer serialize() {
+ return ByteBuffer.wrap(toProtocol().toByteArray());
+ }
+
+ /**
+ * Parse the serialized form of this protocol message.
+ *
+ * <p>Intended to help interoperability by allowing Flight clients to obtain stream info from non-Flight services.
+ *
+ * @param serialized The serialized form of the FlightDescriptor, as returned by {@link #serialize()}.
+ * @return The deserialized FlightDescriptor.
+ * @throws IOException if the serialized form is invalid.
+ */
+ public static FlightDescriptor deserialize(ByteBuffer serialized) throws IOException {
+ return new FlightDescriptor(Flight.FlightDescriptor.parseFrom(serialized));
+ }
+
@Override
public String toString() {
if (isCmd) {
@@ -135,7 +160,7 @@ public class FlightDescriptor {
if (other.cmd != null) {
return false;
}
- } else if (!cmd.equals(other.cmd)) {
+ } else if (!Arrays.equals(cmd, other.cmd)) {
return false;
}
if (isCmd != other.isCmd) {
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/FlightEndpoint.java b/java/flight/src/main/java/org/apache/arrow/flight/FlightEndpoint.java
index 6f986ce..49ed525 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/FlightEndpoint.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/FlightEndpoint.java
@@ -20,6 +20,7 @@ package org.apache.arrow.flight;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import org.apache.arrow.flight.impl.Flight;
@@ -40,6 +41,7 @@ public class FlightEndpoint {
*/
public FlightEndpoint(Ticket ticket, Location... locations) {
super();
+ Objects.requireNonNull(ticket);
this.locations = ImmutableList.copyOf(locations);
this.ticket = ticket;
}
@@ -47,7 +49,7 @@ public class FlightEndpoint {
/**
* Constructs from the protocol buffer representation.
*/
- public FlightEndpoint(Flight.FlightEndpoint flt) throws URISyntaxException {
+ FlightEndpoint(Flight.FlightEndpoint flt) throws URISyntaxException {
locations = new ArrayList<>();
for (final Flight.Location location : flt.getLocationList()) {
locations.add(new Location(location.getUri()));
@@ -75,4 +77,30 @@ public class FlightEndpoint {
}
return b.build();
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FlightEndpoint that = (FlightEndpoint) o;
+ return locations.equals(that.locations) &&
+ ticket.equals(that.ticket);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(locations, ticket);
+ }
+
+ @Override
+ public String toString() {
+ return "FlightEndpoint{" +
+ "locations=" + locations +
+ ", ticket=" + ticket +
+ '}';
+ }
}
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/FlightInfo.java b/java/flight/src/main/java/org/apache/arrow/flight/FlightInfo.java
index 4159e46..3e2b055 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/FlightInfo.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/FlightInfo.java
@@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.stream.Collectors;
import org.apache.arrow.flight.impl.Flight;
@@ -59,6 +60,9 @@ public class FlightInfo {
public FlightInfo(Schema schema, FlightDescriptor descriptor, List<FlightEndpoint> endpoints, long bytes,
long records) {
super();
+ Objects.requireNonNull(schema);
+ Objects.requireNonNull(descriptor);
+ Objects.requireNonNull(endpoints);
this.schema = schema;
this.descriptor = descriptor;
this.endpoints = endpoints;
@@ -126,6 +130,60 @@ public class FlightInfo {
.setTotalBytes(FlightInfo.this.bytes)
.setTotalRecords(records)
.build();
+ }
+
+ /**
+ * Get the serialized form of this protocol message.
+ *
+ * <p>Intended to help interoperability by allowing non-Flight services to still return Flight types.
+ */
+ public ByteBuffer serialize() {
+ return ByteBuffer.wrap(toProtocol().toByteArray());
+ }
+
+ /**
+ * Parse the serialized form of this protocol message.
+ *
+ * <p>Intended to help interoperability by allowing Flight clients to obtain stream info from non-Flight services.
+ *
+ * @param serialized The serialized form of the FlightInfo, as returned by {@link #serialize()}.
+ * @return The deserialized FlightInfo.
+ * @throws IOException if the serialized form is invalid.
+ * @throws URISyntaxException if the serialized form contains an unsupported URI format.
+ */
+ public static FlightInfo deserialize(ByteBuffer serialized) throws IOException, URISyntaxException {
+ return new FlightInfo(Flight.FlightInfo.parseFrom(serialized));
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FlightInfo that = (FlightInfo) o;
+ return bytes == that.bytes &&
+ records == that.records &&
+ schema.equals(that.schema) &&
+ descriptor.equals(that.descriptor) &&
+ endpoints.equals(that.endpoints);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(schema, descriptor, endpoints, bytes, records);
+ }
+ @Override
+ public String toString() {
+ return "FlightInfo{" +
+ "schema=" + schema +
+ ", descriptor=" + descriptor +
+ ", endpoints=" + endpoints +
+ ", bytes=" + bytes +
+ ", records=" + records +
+ '}';
}
}
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/Location.java b/java/flight/src/main/java/org/apache/arrow/flight/Location.java
index 32556a2..1fbec7b 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/Location.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/Location.java
@@ -22,6 +22,7 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.Objects;
import org.apache.arrow.flight.impl.Flight;
@@ -46,6 +47,7 @@ public class Location {
*/
public Location(URI uri) {
super();
+ Objects.requireNonNull(uri);
this.uri = uri;
}
@@ -87,7 +89,7 @@ public class Location {
/**
* Convert this Location into its protocol-level representation.
*/
- public Flight.Location toProtocol() {
+ Flight.Location toProtocol() {
return Flight.Location.newBuilder().setUri(uri.toString()).build();
}
@@ -129,4 +131,28 @@ public class Location {
throw new IllegalArgumentException(e);
}
}
+
+ @Override
+ public String toString() {
+ return "Location{" +
+ "uri=" + uri +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Location location = (Location) o;
+ return uri.equals(location.uri);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(uri);
+ }
}
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/Result.java b/java/flight/src/main/java/org/apache/arrow/flight/Result.java
index bd3ad4e..5d6ce48 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/Result.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/Result.java
@@ -34,7 +34,7 @@ public class Result {
this.body = body;
}
- public Result(Flight.Result result) {
+ Result(Flight.Result result) {
this.body = result.getBody().toByteArray();
}
diff --git a/java/flight/src/main/java/org/apache/arrow/flight/Ticket.java b/java/flight/src/main/java/org/apache/arrow/flight/Ticket.java
index 4c39d8c..a93cd08 100644
--- a/java/flight/src/main/java/org/apache/arrow/flight/Ticket.java
+++ b/java/flight/src/main/java/org/apache/arrow/flight/Ticket.java
@@ -17,6 +17,8 @@
package org.apache.arrow.flight;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import org.apache.arrow.flight.impl.Flight;
@@ -48,6 +50,28 @@ public class Ticket {
.build();
}
+ /**
+ * Get the serialized form of this protocol message.
+ *
+ * <p>Intended to help interoperability by allowing non-Flight services to still return Flight types.
+ */
+ public ByteBuffer serialize() {
+ return ByteBuffer.wrap(toProtocol().toByteArray());
+ }
+
+ /**
+ * Parse the serialized form of this protocol message.
+ *
+ * <p>Intended to help interoperability by allowing Flight clients to obtain stream info from non-Flight services.
+ *
+ * @param serialized The serialized form of the Ticket, as returned by {@link #serialize()}.
+ * @return The deserialized Ticket.
+ * @throws IOException if the serialized form is invalid.
+ */
+ public static Ticket deserialize(ByteBuffer serialized) throws IOException {
+ return new Ticket(Flight.Ticket.parseFrom(serialized));
+ }
+
@Override
public int hashCode() {
final int prime = 31;
diff --git a/java/flight/src/test/java/org/apache/arrow/flight/TestBasicOperation.java b/java/flight/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
index 9a2c5e3..c764502 100644
--- a/java/flight/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
+++ b/java/flight/src/test/java/org/apache/arrow/flight/TestBasicOperation.java
@@ -19,7 +19,12 @@ package org.apache.arrow.flight;
import java.net.URI;
import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
+import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -30,6 +35,9 @@ import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.IntVector;
import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
import org.junit.Assert;
import org.junit.Test;
@@ -63,6 +71,45 @@ public class TestBasicOperation {
}
@Test
+ public void roundTripTicket() throws Exception {
+ final Ticket ticket = new Ticket(new byte[]{0, 1, 2, 3, 4, 5});
+ Assert.assertEquals(ticket, Ticket.deserialize(ticket.serialize()));
+ }
+
+ @Test
+ public void roundTripInfo() throws Exception {
+ final Map<String, String> metadata = new HashMap<>();
+ metadata.put("foo", "bar");
+ final Schema schema = new Schema(Arrays.asList(
+ Field.nullable("a", new ArrowType.Int(32, true)),
+ Field.nullable("b", new ArrowType.FixedSizeBinary(32))
+ ), metadata);
+ final FlightInfo info1 = new FlightInfo(schema, FlightDescriptor.path(), Collections.emptyList(), -1, -1);
+ final FlightInfo info2 = new FlightInfo(schema, FlightDescriptor.command(new byte[2]),
+ Collections.singletonList(new FlightEndpoint(
+ new Ticket(new byte[10]), Location.forGrpcDomainSocket("/tmp/test.sock"))), 200, 500);
+ final FlightInfo info3 = new FlightInfo(schema, FlightDescriptor.path("a", "b"),
+ Arrays.asList(new FlightEndpoint(
+ new Ticket(new byte[10]), Location.forGrpcDomainSocket("/tmp/test.sock")),
+ new FlightEndpoint(
+ new Ticket(new byte[10]), Location.forGrpcDomainSocket("/tmp/test.sock"),
+ Location.forGrpcInsecure("localhost", 50051))
+ ), 200, 500);
+
+ Assert.assertEquals(info1, FlightInfo.deserialize(info1.serialize()));
+ Assert.assertEquals(info2, FlightInfo.deserialize(info2.serialize()));
+ Assert.assertEquals(info3, FlightInfo.deserialize(info3.serialize()));
+ }
+
+ @Test
+ public void roundTripDescriptor() throws Exception {
+ final FlightDescriptor cmd = FlightDescriptor.command("test command".getBytes(StandardCharsets.UTF_8));
+ Assert.assertEquals(cmd, FlightDescriptor.deserialize(cmd.serialize()));
+ final FlightDescriptor path = FlightDescriptor.path("foo", "bar", "test.arrow");
+ Assert.assertEquals(path, FlightDescriptor.deserialize(path.serialize()));
+ }
+
+ @Test
public void getDescriptors() throws Exception {
test(c -> {
for (FlightInfo i : c.listFlights(Criteria.ALL)) {
diff --git a/python/pyarrow/_flight.pyx b/python/pyarrow/_flight.pyx
index 5bfc88a..8c51c14 100644
--- a/python/pyarrow/_flight.pyx
+++ b/python/pyarrow/_flight.pyx
@@ -223,7 +223,12 @@ cdef class FlightDescriptor:
return self.descriptor.path
def __repr__(self):
- return "<FlightDescriptor type: {!r}>".format(self.descriptor_type)
+ if self.descriptor_type == DescriptorType.PATH:
+ return "<FlightDescriptor path: {!r}>".format(self.path)
+ elif self.descriptor_type == DescriptorType.CMD:
+ return "<FlightDescriptor command: {!r}>".format(self.command)
+ else:
+ return "<FlightDescriptor type: {!r}>".format(self.descriptor_type)
@staticmethod
cdef CFlightDescriptor unwrap(descriptor) except *:
@@ -232,6 +237,34 @@ cdef class FlightDescriptor:
type(descriptor)))
return (<FlightDescriptor> descriptor).descriptor
+ def serialize(self):
+ """Get the wire-format representation of this type.
+
+ Useful when interoperating with non-Flight systems (e.g. REST
+ services) that may want to return Flight types.
+
+ """
+ cdef c_string out
+ check_status(self.descriptor.SerializeToString(&out))
+ return out
+
+ @classmethod
+ def deserialize(cls, serialized):
+ """Parse the wire-format representation of this type.
+
+ Useful when interoperating with non-Flight systems (e.g. REST
+ services) that may want to return Flight types.
+
+ """
+ cdef FlightDescriptor descriptor = \
+ FlightDescriptor.__new__(FlightDescriptor)
+ check_status(CFlightDescriptor.Deserialize(
+ tobytes(serialized), &descriptor.descriptor))
+ return descriptor
+
+ def __eq__(self, FlightDescriptor other):
+ return self.descriptor == other.descriptor
+
cdef class Ticket:
"""A ticket for requesting a Flight stream."""
@@ -246,6 +279,36 @@ cdef class Ticket:
def ticket(self):
return self.ticket.ticket
+ def serialize(self):
+ """Get the wire-format representation of this type.
+
+ Useful when interoperating with non-Flight systems (e.g. REST
+ services) that may want to return Flight types.
+
+ """
+ cdef c_string out
+ check_status(self.ticket.SerializeToString(&out))
+ return out
+
+ @classmethod
+ def deserialize(cls, serialized):
+ """Parse the wire-format representation of this type.
+
+ Useful when interoperating with non-Flight systems (e.g. REST
+ services) that may want to return Flight types.
+
+ """
+ cdef:
+ CTicket c_ticket
+ Ticket ticket
+ check_status(CTicket.Deserialize(tobytes(serialized), &c_ticket))
+ ticket = Ticket.__new__(Ticket)
+ ticket.ticket = c_ticket
+ return ticket
+
+ def __eq__(self, Ticket other):
+ return self.ticket == other.ticket
+
def __repr__(self):
return '<Ticket {}>'.format(self.ticket.ticket)
@@ -366,6 +429,13 @@ cdef class FlightEndpoint:
return [Location.wrap(location)
for location in self.endpoint.locations]
+ def __repr__(self):
+ return "<FlightEndpoint ticket: {!r} locations: {!r}>".format(
+ self.ticket, self.locations)
+
+ def __eq__(self, FlightEndpoint other):
+ return self.endpoint == other.endpoint
+
cdef class FlightInfo:
"""A description of a Flight stream."""
@@ -449,6 +519,30 @@ cdef class FlightInfo:
result.append(py_endpoint)
return result
+ def serialize(self):
+ """Get the wire-format representation of this type.
+
+ Useful when interoperating with non-Flight systems (e.g. REST
+ services) that may want to return Flight types.
+
+ """
+ cdef c_string out
+ check_status(self.info.get().SerializeToString(&out))
+ return out
+
+ @classmethod
+ def deserialize(cls, serialized):
+ """Parse the wire-format representation of this type.
+
+ Useful when interoperating with non-Flight systems (e.g. REST
+ services) that may want to return Flight types.
+
+ """
+ cdef FlightInfo info = FlightInfo.__new__(FlightInfo)
+ check_status(CFlightInfo.Deserialize(
+ tobytes(serialized), &info.info))
+ return info
+
cdef class FlightStreamChunk:
"""A RecordBatch with application metadata on the side."""
diff --git a/python/pyarrow/includes/libarrow_flight.pxd b/python/pyarrow/includes/libarrow_flight.pxd
index ed0b33e..7ac744a 100644
--- a/python/pyarrow/includes/libarrow_flight.pxd
+++ b/python/pyarrow/includes/libarrow_flight.pxd
@@ -69,10 +69,19 @@ cdef extern from "arrow/flight/api.h" namespace "arrow" nogil:
CDescriptorType type
c_string cmd
vector[c_string] path
+ CStatus SerializeToString(c_string* out)
+ @staticmethod
+ CStatus Deserialize(const c_string& serialized,
+ CFlightDescriptor* out)
+ bint operator==(CFlightDescriptor)
cdef cppclass CTicket" arrow::flight::Ticket":
CTicket()
c_string ticket
+ bint operator==(CTicket)
+ CStatus SerializeToString(c_string* out)
+ @staticmethod
+ CStatus Deserialize(const c_string& serialized, CTicket* out)
cdef cppclass CCriteria" arrow::flight::Criteria":
CCriteria()
@@ -98,6 +107,8 @@ cdef extern from "arrow/flight/api.h" namespace "arrow" nogil:
CTicket ticket
vector[CLocation] locations
+ bint operator==(CFlightEndpoint)
+
cdef cppclass CFlightInfo" arrow::flight::FlightInfo":
CFlightInfo(CFlightInfo info)
int64_t total_records()
@@ -105,6 +116,10 @@ cdef extern from "arrow/flight/api.h" namespace "arrow" nogil:
CStatus GetSchema(CDictionaryMemo* memo, shared_ptr[CSchema]* out)
CFlightDescriptor& descriptor()
const vector[CFlightEndpoint]& endpoints()
+ CStatus SerializeToString(c_string* out)
+ @staticmethod
+ CStatus Deserialize(const c_string& serialized,
+ unique_ptr[CFlightInfo]* out)
cdef cppclass CFlightListing" arrow::flight::FlightListing":
CStatus Next(unique_ptr[CFlightInfo]* info)
diff --git a/python/pyarrow/tests/test_flight.py b/python/pyarrow/tests/test_flight.py
index 2912e29..ec8c52c 100644
--- a/python/pyarrow/tests/test_flight.py
+++ b/python/pyarrow/tests/test_flight.py
@@ -783,3 +783,35 @@ def test_cancel_do_get_threaded():
with result_lock:
assert raised_proper_exception.is_set()
+
+
+def test_roundtrip_types():
+ """Make sure serializable types round-trip."""
+ ticket = flight.Ticket("foo")
+ assert ticket == flight.Ticket.deserialize(ticket.serialize())
+
+ desc = flight.FlightDescriptor.for_command("test")
+ assert desc == flight.FlightDescriptor.deserialize(desc.serialize())
+
+ desc = flight.FlightDescriptor.for_path("a", "b", "test.arrow")
+ assert desc == flight.FlightDescriptor.deserialize(desc.serialize())
+
+ info = flight.FlightInfo(
+ pa.schema([('a', pa.int32())]),
+ desc,
+ [
+ flight.FlightEndpoint(b'', ['grpc://test']),
+ flight.FlightEndpoint(
+ b'',
+ [flight.Location.for_grpc_tcp('localhost', 5005)],
+ ),
+ ],
+ -1,
+ -1,
+ )
+ info2 = flight.FlightInfo.deserialize(info.serialize())
+ assert info.schema == info2.schema
+ assert info.descriptor == info2.descriptor
+ assert info.total_bytes == info2.total_bytes
+ assert info.total_records == info2.total_records
+ assert info.endpoints == info2.endpoints