alamb commented on code in PR #10432:
URL: https://github.com/apache/datafusion/pull/10432#discussion_r1596654105
##########
datafusion-examples/examples/flight/flight_sql_server.rs:
##########
@@ -304,267 +313,50 @@ impl FlightSqlService for FlightSqlServiceImpl {
};
let buf = fetch.as_any().encode_to_vec().into();
let ticket = Ticket { ticket: buf };
- let endpoint = FlightEndpoint {
- ticket: Some(ticket),
- location: vec![],
- expiration_time: None,
- app_metadata: Default::default(),
- };
- let endpoints = vec![endpoint];
-
- let message = SchemaAsIpc::new(&schema, &IpcWriteOptions::default())
- .try_into()
- .map_err(|e| status!("Unable to serialize schema", e))?;
- let IpcMessage(schema_bytes) = message;
- let flight_desc = FlightDescriptor {
- r#type: DescriptorType::Cmd.into(),
- cmd: Default::default(),
- path: vec![],
- };
- // send -1 for total_records and total_bytes instead of iterating over
all the
- // batches to get num_rows() and total byte size.
- let info = FlightInfo {
- schema: schema_bytes,
- flight_descriptor: Some(flight_desc),
- endpoint: endpoints,
- total_records: -1_i64,
- total_bytes: -1_i64,
- ordered: false,
- app_metadata: Default::default(),
- };
+ let info = FlightInfo::new()
+ // Encode the Arrow schema
+ .try_with_schema(&schema)
+ .expect("encoding failed")
+ .with_endpoint(FlightEndpoint::new().with_ticket(ticket))
+ .with_descriptor(FlightDescriptor {
+ r#type: DescriptorType::Cmd.into(),
+ cmd: Default::default(),
+ path: vec![],
+ });
let resp = Response::new(info);
Ok(resp)
}
- async fn get_flight_info_catalogs(
- &self,
- _query: CommandGetCatalogs,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_catalogs");
- Err(Status::unimplemented("Implement get_flight_info_catalogs"))
- }
-
- async fn get_flight_info_schemas(
- &self,
- _query: CommandGetDbSchemas,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_schemas");
- Err(Status::unimplemented("Implement get_flight_info_schemas"))
- }
-
async fn get_flight_info_tables(
&self,
_query: CommandGetTables,
- _request: Request<FlightDescriptor>,
+ request: Request<FlightDescriptor>,
) -> Result<Response<FlightInfo>, Status> {
info!("get_flight_info_tables");
- Err(Status::unimplemented("Implement get_flight_info_tables"))
- }
-
- async fn get_flight_info_table_types(
- &self,
- _query: CommandGetTableTypes,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_table_types");
- Err(Status::unimplemented(
- "Implement get_flight_info_table_types",
- ))
- }
-
- async fn get_flight_info_sql_info(
- &self,
- _query: CommandGetSqlInfo,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_sql_info");
- Err(Status::unimplemented("Implement CommandGetSqlInfo"))
- }
-
- async fn get_flight_info_primary_keys(
- &self,
- _query: CommandGetPrimaryKeys,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_primary_keys");
- Err(Status::unimplemented(
- "Implement get_flight_info_primary_keys",
- ))
- }
-
- async fn get_flight_info_exported_keys(
- &self,
- _query: CommandGetExportedKeys,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_exported_keys");
- Err(Status::unimplemented(
- "Implement get_flight_info_exported_keys",
- ))
- }
-
- async fn get_flight_info_imported_keys(
- &self,
- _query: CommandGetImportedKeys,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_imported_keys");
- Err(Status::unimplemented(
- "Implement get_flight_info_imported_keys",
- ))
- }
-
- async fn get_flight_info_cross_reference(
- &self,
- _query: CommandGetCrossReference,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_cross_reference");
- Err(Status::unimplemented(
- "Implement get_flight_info_cross_reference",
- ))
- }
-
- async fn get_flight_info_xdbc_type_info(
- &self,
- _query: CommandGetXdbcTypeInfo,
- _request: Request<FlightDescriptor>,
- ) -> Result<Response<FlightInfo>, Status> {
- info!("get_flight_info_xdbc_type_info");
- Err(Status::unimplemented(
- "Implement get_flight_info_xdbc_type_info",
- ))
- }
-
- async fn do_get_statement(
- &self,
- _ticket: TicketStatementQuery,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_statement");
- Err(Status::unimplemented("Implement do_get_statement"))
- }
-
- async fn do_get_prepared_statement(
- &self,
- _query: CommandPreparedStatementQuery,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_prepared_statement");
- Err(Status::unimplemented("Implement do_get_prepared_statement"))
- }
-
- async fn do_get_catalogs(
- &self,
- _query: CommandGetCatalogs,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_catalogs");
- Err(Status::unimplemented("Implement do_get_catalogs"))
- }
-
- async fn do_get_schemas(
- &self,
- _query: CommandGetDbSchemas,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_schemas");
- Err(Status::unimplemented("Implement do_get_schemas"))
- }
-
- async fn do_get_tables(
- &self,
- _query: CommandGetTables,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_tables");
- Err(Status::unimplemented("Implement do_get_tables"))
- }
-
- async fn do_get_table_types(
- &self,
- _query: CommandGetTableTypes,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_table_types");
- Err(Status::unimplemented("Implement do_get_table_types"))
- }
-
- async fn do_get_sql_info(
- &self,
- _query: CommandGetSqlInfo,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_sql_info");
- Err(Status::unimplemented("Implement do_get_sql_info"))
- }
-
- async fn do_get_primary_keys(
- &self,
- _query: CommandGetPrimaryKeys,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_primary_keys");
- Err(Status::unimplemented("Implement do_get_primary_keys"))
- }
-
- async fn do_get_exported_keys(
- &self,
- _query: CommandGetExportedKeys,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_exported_keys");
- Err(Status::unimplemented("Implement do_get_exported_keys"))
- }
-
- async fn do_get_imported_keys(
- &self,
- _query: CommandGetImportedKeys,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_imported_keys");
- Err(Status::unimplemented("Implement do_get_imported_keys"))
- }
-
- async fn do_get_cross_reference(
- &self,
- _query: CommandGetCrossReference,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_cross_reference");
- Err(Status::unimplemented("Implement do_get_cross_reference"))
- }
+ let ctx = self.get_ctx(&request)?;
+ let data = self.tables(ctx).await;
+ let schema = data.schema();
- async fn do_get_xdbc_type_info(
- &self,
- _query: CommandGetXdbcTypeInfo,
- _request: Request<Ticket>,
- ) -> Result<Response<<Self as FlightService>::DoGetStream>, Status> {
- info!("do_get_xdbc_type_info");
- Err(Status::unimplemented("Implement do_get_xdbc_type_info"))
- }
+ let uuid = Uuid::new_v4().hyphenated().to_string();
+ self.results.insert(uuid.clone(), vec![data]);
- async fn do_put_statement_update(
- &self,
- _ticket: CommandStatementUpdate,
- _request: Request<PeekableFlightDataStream>,
- ) -> Result<i64, Status> {
- info!("do_put_statement_update");
- Err(Status::unimplemented("Implement do_put_statement_update"))
- }
+ let fetch = FetchResults { handle: uuid };
+ let buf = fetch.as_any().encode_to_vec().into();
+ let ticket = Ticket { ticket: buf };
- async fn do_put_prepared_statement_query(
- &self,
- _query: CommandPreparedStatementQuery,
- _request: Request<PeekableFlightDataStream>,
- ) -> Result<Response<<Self as FlightService>::DoPutStream>, Status> {
- info!("do_put_prepared_statement_query");
- Err(Status::unimplemented(
- "Implement do_put_prepared_statement_query",
- ))
+ let info = FlightInfo::new()
Review Comment:
😍
--
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-unsubscr...@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscr...@datafusion.apache.org
For additional commands, e-mail: github-h...@datafusion.apache.org
