You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "ccciudatu (via GitHub)" <gi...@apache.org> on 2024/08/11 14:32:38 UTC
[PR] Generic Arrow Flight RPC data source with a Flight SQL adapter [datafusion]
ccciudatu opened a new pull request, #11938:
URL: https://github.com/apache/datafusion/pull/11938
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #7731.
## Rationale for this change
Create external tables backed by Arrow Flight(SQL) services.
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
- A generic `FlightDataSource` that can integrate any Arrow Flight RPC service as a `TableProviderFactory` datasource, expecting a `FlightSourceAdapter` implementation to handle the `GetFlightInfo` flow.
- Default `FlightSourceAdapter` implementation for Flight SQL, registered as `FLIGHT_SQL`.
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
yes
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2287605620
I added the proto marshalling and marked the PR "ready for review", once again.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1717608207
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
https://github.com/apache/arrow-rs/pull/6254
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2297439748
Moved to its own repo: https://github.com/datafusion-contrib/datafusion-table-provider-flight/pull/1
Thanks, @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.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1717601113
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
https://github.com/apache/arrow-rs/issues/6253
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1717586777
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
TODO: Remove this "stolen" client implementation and submit a PR to arrow-rs to make the token visible in the official client.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1717618646
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
@alamb on average, my PRs are of reasonable size. :)
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-cli] Generic Arrow Flight RPC data source with a Flight SQL adapter [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2286284151
@alamb The PR title is misleading; this is *not* meant to be a `datafusion-cli` patch. It only got renamed because I added a follow-up commit to make the new `FLIGHT_SQL` table "storage" type work with `datafusion-cli` (i.e. register the new config namespace).
However, I'm glad this triggered such an interesting high-level discussion about revisiting the cli package. :)
I will squash the whole thing in a single commit and restore the original title .
Also, I just realized that I need to add proto serialization for the new FlightExec physical plan, so I'll convert this the PR to a draft for now.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1716654822
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -1081,6 +1091,67 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
sort_order,
)))
}
+ #[cfg(feature = "flight")]
+ PhysicalPlanType::FlightScan(flight_scan) => {
+ let partitions = flight_scan
+ .partitions
+ .iter()
+ .map(|partition_node| {
+ FlightPartition::restore(
+ partition_node.locations.clone(),
+ partition_node.token.clone(),
+ )
+ })
+ .collect();
+ let plan_props =
+ flight_scan.plan_properties.clone().ok_or_else(|| {
+ DataFusionError::Internal(
+ "Missing plan_properties in FlightExec".into(),
+ )
+ })?;
+ let schema: Arc<Schema> = Arc::new(convert_required!(plan_props.schema)?);
+ let codec = DefaultPhysicalExtensionCodec {};
+ let partitioning = parse_protobuf_partitioning(
+ plan_props.partitioning.as_ref(),
+ registry,
+ schema.as_ref(),
+ &codec,
+ )?
+ .ok_or_else(|| {
+ DataFusionError::Internal(
+ "Missing partitioning from plan properties".into(),
+ )
+ })?;
+ let orderings = &plan_props
+ .output_ordering
+ .iter()
+ .map(|node_collection| {
+ parse_physical_sort_exprs(
+ &node_collection.physical_sort_expr_nodes,
+ registry,
+ &schema,
+ &codec,
+ )
+ .unwrap()
+ })
+ .collect::<Vec<_>>();
+ let eq_props =
+ EquivalenceProperties::new_with_orderings(schema, orderings);
+ let execution_mode = match plan_props.execution_mode {
+ 0 => ExecutionMode::Bounded,
+ 1 => ExecutionMode::Unbounded,
+ 2 => ExecutionMode::PipelineBreaking,
+ _ => unreachable!("Unexpected execution mode"),
+ };
+ let plan_properties =
+ PlanProperties::new(eq_props, partitioning, execution_mode);
+
+ Ok(Arc::new(FlightExec::restore(
+ partitions,
+ plan_properties,
+ &flight_scan.grpc_headers,
+ )))
Review Comment:
fixed
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1720105929
##########
datafusion/core/src/datasource/flight/config.rs:
##########
@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Only meant for registering the `flight` namespace with `datafusion-cli`
+
+use datafusion_common::config::{ConfigEntry, ConfigExtension, ExtensionOptions};
+use std::any::Any;
+use std::collections::HashMap;
+
+/// Collects and reports back config entries. Only used to persuade `datafusion-cli`
+/// to accept the `flight.` prefix for `CREATE EXTERNAL TABLE` options.
+#[derive(Default, Debug, Clone)]
+pub struct FlightOptions {
Review Comment:
The configuration system is another thing I think that a focused standalone binary would be able to do better than DataFusion's key=value system
##########
datafusion/core/src/datasource/flight/mod.rs:
##########
@@ -0,0 +1,259 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Generic [FlightTableFactory] that can connect to Arrow Flight services,
+//! with a [sql::FlightSqlDriver] provided out-of-the-box.
+
+use std::any::Any;
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow_flight::error::FlightError;
+use arrow_flight::FlightInfo;
+use arrow_schema::SchemaRef;
+use async_trait::async_trait;
+use tonic::metadata::MetadataMap;
+use tonic::transport::Channel;
+
+use datafusion_catalog::{Session, TableProvider, TableProviderFactory};
+use datafusion_common::{project_schema, DataFusionError};
+use datafusion_expr::{CreateExternalTable, Expr, TableType};
+use datafusion_physical_expr::EquivalenceProperties;
+use datafusion_physical_expr::Partitioning::UnknownPartitioning;
+use datafusion_physical_plan::{ExecutionMode, ExecutionPlan, PlanProperties};
+
+use crate::datasource::physical_plan::FlightExec;
+
+pub mod config;
+pub mod sql;
+
+/// Generic Arrow Flight data source. Requires a [FlightDriver] that allows implementors
+/// to integrate any custom Flight RPC service by producing a [FlightMetadata] for some DDL.
+///
+/// # Sample usage:
+/// ```
+/// use std::collections::HashMap;
+/// use arrow_flight::{FlightClient, FlightDescriptor};
+/// use tonic::transport::Channel;
+/// use datafusion::datasource::flight::{FlightMetadata, FlightDriver};
+/// use datafusion::prelude::SessionContext;
+/// use std::sync::Arc;
+/// use datafusion::datasource::flight::FlightTableFactory;
+///
+/// #[derive(Debug, Clone, Default)]
+/// struct CustomFlightDriver {}
+/// #[async_trait::async_trait]
+/// impl FlightDriver for CustomFlightDriver {
+/// async fn metadata(&self, channel: Channel, opts: &HashMap<String, String>)
+/// -> arrow_flight::error::Result<FlightMetadata> {
+/// let mut client = FlightClient::new(channel);
+/// // the `flight.` prefix is an already registered namespace in datafusion-cli
+/// let descriptor = FlightDescriptor::new_cmd(opts["flight.command"].clone());
+/// let flight_info = client.get_flight_info(descriptor).await?;
+/// FlightMetadata::try_from(flight_info)
+/// }
+/// }
+///
+/// #[tokio::main]
+/// async fn main() -> datafusion_common::Result<()> {
+/// let ctx = SessionContext::new();
+/// ctx.state_ref().write().table_factories_mut()
+/// .insert("CUSTOM_FLIGHT".into(), Arc::new(FlightTableFactory::new(
+/// Arc::new(CustomFlightDriver::default())
+/// )));
+/// let _ = ctx.sql(r#"
+/// CREATE EXTERNAL TABLE custom_flight_table STORED AS CUSTOM_FLIGHT
+/// LOCATION 'https://custom.flight.rpc'
+/// OPTIONS ('flight.command' 'select * from everywhere')
+/// "#).await; // will fail as it can't connect to the bogus URL, but we ignore the error
+/// Ok(())
+/// }
+///
+/// ```
+#[derive(Clone, Debug)]
+pub struct FlightTableFactory {
+ driver: Arc<dyn FlightDriver>,
+}
+
+impl FlightTableFactory {
+ /// Create a data source using the provided driver
+ pub fn new(driver: Arc<dyn FlightDriver>) -> Self {
+ Self { driver }
+ }
+
+ /// Convenient way to create a [FlightTable] programatically, as an alternative to DDL.
+ pub async fn open_table(
+ &self,
+ entry_point: impl Into<String>,
+ options: HashMap<String, String>,
+ ) -> datafusion_common::Result<FlightTable> {
+ let origin = entry_point.into();
+ let channel = Channel::from_shared(origin.clone())
+ .unwrap()
+ .connect()
+ .await
+ .map_err(|e| DataFusionError::External(Box::new(e)))?;
+ let metadata = self
+ .driver
+ .metadata(channel.clone(), &options)
+ .await
+ .map_err(|e| DataFusionError::External(Box::new(e)))?;
+ let logical_schema = metadata.plan_properties.eq_properties.schema().clone();
+ Ok(FlightTable {
+ driver: self.driver.clone(),
+ channel,
+ options,
+ origin,
+ logical_schema,
+ })
+ }
+}
+
+#[async_trait]
+impl TableProviderFactory for FlightTableFactory {
+ async fn create(
+ &self,
+ _state: &dyn Session,
+ cmd: &CreateExternalTable,
+ ) -> datafusion_common::Result<Arc<dyn TableProvider>> {
+ let table = self.open_table(&cmd.location, cmd.options.clone()).await?;
+ Ok(Arc::new(table))
+ }
+}
+
+/// Extension point for integrating any Flight RPC service as a [FlightTableFactory].
+/// Handles the initial `GetFlightInfo` call and all its prerequisites (such as `Handshake`),
+/// to produce a [FlightMetadata].
+#[async_trait]
+pub trait FlightDriver: Sync + Send + Debug {
+ /// Returns a [FlightMetadata] from the specified channel,
+ /// according to the provided table options.
+ /// The driver must provide at least a [FlightInfo] in order to construct a flight metadata.
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> arrow_flight::error::Result<FlightMetadata>;
+}
+
+/// The information that a [FlightDriver] must produce
+/// in order to register flights as DataFusion tables.
+#[derive(Clone, Debug)]
+pub struct FlightMetadata {
+ /// FlightInfo object produced by the driver
+ pub(super) flight_info: Arc<FlightInfo>,
+ /// Physical plan properties. Sensible defaults will be used if the
+ /// driver doesn't need (or care) to customize the execution plan.
+ pub(super) plan_properties: Arc<PlanProperties>,
+ /// The gRPC headers to use on the `DoGet` calls
+ pub(super) grpc_metadata: Arc<MetadataMap>,
+}
+
+impl FlightMetadata {
+ /// Provide custom [PlanProperties] to account for service specifics,
+ /// such as known partitioning scheme, unbounded execution mode etc.
+ pub fn new(info: FlightInfo, props: PlanProperties, grpc: MetadataMap) -> Self {
Review Comment:
this is really cool -- we actually implemented some version of this custom for InfluxDB 3.0 for services that exchange data between themselves
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
Review Comment:
Yeah, the configuration story for coinfiguring datasources like this is sorely lacking in datafusion-cli
It would be kind of cool to support something like
```sql
create crednetials for flightsql as ....
select * from 'flightsql://myhost/SELECT%20*%20FROM%20foo'
```
##########
datafusion/core/src/datasource/flight/mod.rs:
##########
@@ -0,0 +1,259 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Generic [FlightTableFactory] that can connect to Arrow Flight services,
+//! with a [sql::FlightSqlDriver] provided out-of-the-box.
+
+use std::any::Any;
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow_flight::error::FlightError;
+use arrow_flight::FlightInfo;
+use arrow_schema::SchemaRef;
+use async_trait::async_trait;
+use tonic::metadata::MetadataMap;
+use tonic::transport::Channel;
+
+use datafusion_catalog::{Session, TableProvider, TableProviderFactory};
+use datafusion_common::{project_schema, DataFusionError};
+use datafusion_expr::{CreateExternalTable, Expr, TableType};
+use datafusion_physical_expr::EquivalenceProperties;
+use datafusion_physical_expr::Partitioning::UnknownPartitioning;
+use datafusion_physical_plan::{ExecutionMode, ExecutionPlan, PlanProperties};
+
+use crate::datasource::physical_plan::FlightExec;
+
+pub mod config;
+pub mod sql;
+
+/// Generic Arrow Flight data source. Requires a [FlightDriver] that allows implementors
+/// to integrate any custom Flight RPC service by producing a [FlightMetadata] for some DDL.
+///
+/// # Sample usage:
+/// ```
+/// use std::collections::HashMap;
+/// use arrow_flight::{FlightClient, FlightDescriptor};
+/// use tonic::transport::Channel;
+/// use datafusion::datasource::flight::{FlightMetadata, FlightDriver};
+/// use datafusion::prelude::SessionContext;
+/// use std::sync::Arc;
+/// use datafusion::datasource::flight::FlightTableFactory;
+///
+/// #[derive(Debug, Clone, Default)]
+/// struct CustomFlightDriver {}
+/// #[async_trait::async_trait]
+/// impl FlightDriver for CustomFlightDriver {
+/// async fn metadata(&self, channel: Channel, opts: &HashMap<String, String>)
+/// -> arrow_flight::error::Result<FlightMetadata> {
+/// let mut client = FlightClient::new(channel);
+/// // the `flight.` prefix is an already registered namespace in datafusion-cli
+/// let descriptor = FlightDescriptor::new_cmd(opts["flight.command"].clone());
+/// let flight_info = client.get_flight_info(descriptor).await?;
+/// FlightMetadata::try_from(flight_info)
+/// }
+/// }
+///
+/// #[tokio::main]
+/// async fn main() -> datafusion_common::Result<()> {
+/// let ctx = SessionContext::new();
+/// ctx.state_ref().write().table_factories_mut()
+/// .insert("CUSTOM_FLIGHT".into(), Arc::new(FlightTableFactory::new(
+/// Arc::new(CustomFlightDriver::default())
+/// )));
+/// let _ = ctx.sql(r#"
Review Comment:
this is really cool
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2296036939
Thank you, @alamb . I will move this to the new repo and close it from here.
It can't really be moved as such, as the proto stuff is cumbersome to replicate outside of this repo, so I think I'll switch to a serde-based codec for the new physical plan node.
The `datafusion-cli` integration was never a priority for me (I just thought it was the civilised way to add new table factories), so I have no problem living without it for now.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2289078249
@alamb Got it, makes sense.
My initial thought was that having this in the main repo would enable a better overall integration (i.e. datafusion-cli support, transparent proto serialization, etc.) but I guess having this in a contrib project should not limit these options. I'll have a better look at the project you linked to see if that's a better fit, but I'll go with whatever you decide if you come to a conclusion.
Separately, I promise I'll stop force-pushing updates now that I know it's being reviewed. I may push some fixups though... :)
The last change was to account for server implementations that do not allow reusing the same FlightInfo object, so now GetFlightInfo is called before every query against a "flight table", rather than only once on table creation.
Also, I found a weird/buggy behaviour related to some schema mismatch, i.e. columns seem to get reordered in certain scenarios. I will try to reproduce that in a test and follow up with the fix or at least the failing test to reproduce 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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1719066241
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
I appreciate it @ccciudatu -- I am sorry I haven't had a chance to comment more thoroughly on this PR yet...
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2288557502
> @alamb The `[datafusion-cli]` commit was just a follow-up to make the new `FLIGHT_SQL` table "storage" type available in `datafusion-cli` (i.e. register the new config namespace). I'm sorry if that was misleading. The patch is supposed to add support for Arrow Flight (SQL) data sources.
FWI I wrote up my thoughts here; https://github.com/apache/datafusion/issues/11979
I hope to review this PR later today
One thing that might be useful to ask is "does this code need to be in the datafusion repo, or could it be put in another repo -- like for example https://github.com/datafusion-contrib/datafusion-table-providers 🤔
We could make you your own repo if you wanted
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1717601113
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
https://github.com/apache/arrow-rs/issues/6253
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] [datafusion-cli] Generic Arrow Flight RPC data source with a Flight SQL adapter [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2284990965
Thank you @ccciudatu
I have been thinking recently that we need to figure out what to do with `datafusion-cli` -- it seems to currently be serving two roles:
1. Debug tool for playing around with supported sql, making reproducers
2. An actual CLI tool with pre-built integrations (I am thinking specifically of https://github.com/apache/datafusion/pull/10792 from @xinlifoobar here)
There is a tension for adding integrations into datafusion-cli as it makes the overall project that much more complicated
@stuartcarnie and I were am thinking of writing up a proposal to make a `dfdb`, or `dfcli` binary where we could start adding all these pre-built integrations. I think that would be quite compelling
I plan to try to write this up / maybe try and make an example repo or something
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1716648030
##########
datafusion/proto/proto/datafusion.proto:
##########
@@ -1199,3 +1200,28 @@ message PartitionStats {
int64 num_bytes = 3;
repeated datafusion_common.ColumnStats column_stats = 4;
}
+
+message FlightScanExecNode {
+ repeated FlightPartitionNode partitions = 1;
+ PlanPropertiesNode plan_properties = 2;
+ map<string, bytes> grpc_headers = 3;
+}
+
+message FlightPartitionNode {
+ repeated string locations = 1;
+ bytes token = 2;
+}
+
+message PlanPropertiesNode {
+ datafusion_common.Schema schema = 1;
+ repeated PhysicalSortExprNodeCollection output_ordering = 2;
+ Partitioning partitioning = 3;
+ ExecutionMode execution_mode = 4;
+}
+
+enum ExecutionMode {
+ Bounded = 0;
+ Unbonded = 1;
Review Comment:
Just noticed this typo :(
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1716651233
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -1081,6 +1091,67 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
sort_order,
)))
}
+ #[cfg(feature = "flight")]
+ PhysicalPlanType::FlightScan(flight_scan) => {
+ let partitions = flight_scan
+ .partitions
+ .iter()
+ .map(|partition_node| {
+ FlightPartition::restore(
+ partition_node.locations.clone(),
+ partition_node.token.clone(),
+ )
+ })
+ .collect();
+ let plan_props =
+ flight_scan.plan_properties.clone().ok_or_else(|| {
+ DataFusionError::Internal(
+ "Missing plan_properties in FlightExec".into(),
+ )
+ })?;
+ let schema: Arc<Schema> = Arc::new(convert_required!(plan_props.schema)?);
+ let codec = DefaultPhysicalExtensionCodec {};
+ let partitioning = parse_protobuf_partitioning(
+ plan_props.partitioning.as_ref(),
+ registry,
+ schema.as_ref(),
+ &codec,
+ )?
+ .ok_or_else(|| {
+ DataFusionError::Internal(
+ "Missing partitioning from plan properties".into(),
+ )
+ })?;
+ let orderings = &plan_props
+ .output_ordering
+ .iter()
+ .map(|node_collection| {
+ parse_physical_sort_exprs(
+ &node_collection.physical_sort_expr_nodes,
+ registry,
+ &schema,
+ &codec,
+ )
+ .unwrap()
+ })
+ .collect::<Vec<_>>();
+ let eq_props =
+ EquivalenceProperties::new_with_orderings(schema, orderings);
+ let execution_mode = match plan_props.execution_mode {
+ 0 => ExecutionMode::Bounded,
+ 1 => ExecutionMode::Unbounded,
+ 2 => ExecutionMode::PipelineBreaking,
+ _ => unreachable!("Unexpected execution mode"),
+ };
+ let plan_properties =
+ PlanProperties::new(eq_props, partitioning, execution_mode);
+
+ Ok(Arc::new(FlightExec::restore(
+ partitions,
+ plan_properties,
+ &flight_scan.grpc_headers,
+ )))
Review Comment:
The feature flag above will fail the build when the "flight" feature is disabled. I'll add an "else" to let it compile and also fix it for the "parquet" feature flag.
However, there are other places where disabling `parquet` will fail the build, but I don't want to expand the scope of this PR.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2311171980
You are right, @phillipleblanc, I enjoyed having my own repo for a while, but I soon started to feel lonely. :)
I'm glad you brought this up! I'll try to follow up with a PR to make this part of [datafusion-table-providers](https://github.com/datafusion-contrib/datafusion-table-providers), where it most likely belongs.
As a side note, I only published it here initially because I noticed that https://github.com/apache/datafusion/issues/7731 was open and acknowledged.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu closed pull request #11938: Generic FlightTableFactory with a default FlightSqlDriver
URL: https://github.com/apache/datafusion/pull/11938
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "phillipleblanc (via GitHub)" <gi...@apache.org>.
phillipleblanc commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2302261267
@ccciudatu Just seeing this now - I'm one of the maintainers of https://github.com/datafusion-contrib/datafusion-table-providers. I think it would be useful to have a single crate of community maintained TableProviders and I think this would fit nicely in there, and I could make you a maintainer on that repo as well.
We've also done work to make the table providers in there play nicely with https://github.com/datafusion-contrib/datafusion-federation, which could also be good for this implementation. Also happy if you want to keep it in a separate crate/repo, we can link to it from README.md.
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1720097023
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
that is nice of you to say -- the whole point of doing this in open source is to get other people to contribute things and a key way to do that is to actually look at their proposals 😆
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on code in PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#discussion_r1719639813
##########
datafusion/core/src/datasource/flight/sql.rs:
##########
@@ -0,0 +1,475 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Default [FlightDriver] for Flight SQL
+
+use std::collections::HashMap;
+use std::str::FromStr;
+
+use arrow_flight::error::Result;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt};
+use arrow_flight::{FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse};
+use arrow_schema::ArrowError;
+use async_trait::async_trait;
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
+use bytes::Bytes;
+use futures::{stream, TryStreamExt};
+use prost::Message;
+use tonic::metadata::{AsciiMetadataKey, MetadataMap};
+use tonic::transport::Channel;
+use tonic::IntoRequest;
+
+use crate::datasource::flight::{FlightDriver, FlightMetadata};
+
+/// Default Flight SQL driver. Requires a `flight.sql.query` to be passed as a table option.
+/// If `flight.sql.username` (and optionally `flight.sql.password`) are passed,
+/// will perform the `Handshake` using basic authentication.
+/// Any additional headers can be passed as table options using the `flight.sql.header.` prefix.
+///
+/// A [crate::datasource::flight::FlightTableFactory] using this driver is registered
+/// with the default `SessionContext` under the name `FLIGHT_SQL`.
+#[derive(Clone, Debug, Default)]
+pub struct FlightSqlDriver {}
+
+#[async_trait]
+impl FlightDriver for FlightSqlDriver {
+ async fn metadata(
+ &self,
+ channel: Channel,
+ options: &HashMap<String, String>,
+ ) -> Result<FlightMetadata> {
+ let mut client = FlightSqlClient::new(channel);
+ let headers = options.iter().filter_map(|(key, value)| {
+ key.strip_prefix("flight.sql.header.")
+ .map(|header_name| (header_name, value))
+ });
+ for header in headers {
+ client.set_header(header.0, header.1)
+ }
+ if let Some(username) = options.get("flight.sql.username") {
+ let default_password = "".to_string();
+ let password = options
+ .get("flight.sql.password")
+ .unwrap_or(&default_password);
+ _ = client.handshake(username, password).await?;
+ }
+ let info = client
+ .execute(options["flight.sql.query"].clone(), None)
+ .await?;
+ let mut grpc_metadata = MetadataMap::new();
+ if let Some(token) = client.token {
+ grpc_metadata.insert(
+ "authorization",
+ format!("Bearer {}", token).parse().unwrap(),
+ );
+ }
+ FlightMetadata::try_new(info, grpc_metadata)
+ }
+}
+
+/////////////////////////////////////////////////////////////////////////
+// Shameless copy/paste from arrow-flight FlightSqlServiceClient
+// This is only needed in order to access the bearer token received
+// during handshake, as the standard client does not expose this information.
+// The bearer token has to be passed to the clients that perform
+// the DoGet operation, since Dremio, Ballista and possibly others
+// expect the bearer token they produce with the handshake response
+// to be set on all subsequent requests, including DoGet.
Review Comment:
No worries, @alamb! The fact that you are even considering reviewing such a huge patch makes you one of the nicest people I know. 🙃
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2297220595
> It can't really be moved as such, as the proto stuff is cumbersome to replicate outside of this repo, so I think I'll switch to a serde-based codec for the new physical plan node.
Makes sense -- thank you for your understanding
--
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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Generic FlightTableFactory with a default FlightSqlDriver [datafusion]
Posted by "ccciudatu (via GitHub)" <gi...@apache.org>.
ccciudatu commented on PR #11938:
URL: https://github.com/apache/datafusion/pull/11938#issuecomment-2289461637
I found my embarrassing mistake and added a fixup 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@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org