Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update datafusion dependency to the latest version #612

Merged
merged 5 commits into from
Jan 30, 2023
Merged

Update datafusion dependency to the latest version #612

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b383870
Update datafusion dependency to the latest version
kyotoYaho Jan 17, 2023
e1be3af
Fix python
kyotoYaho Jan 29, 2023
1a23bb6
Skip ut of test_window_lead due to https://github.com/apache/arrow-da…
kyotoYaho Jan 29, 2023
a728352
Fix clippy
kyotoYaho Jan 29, 2023
80a195f
Merge branch 'main' into issue-610
kyotoYaho Jan 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,13 @@

[workspace]
members = [
"benchmarks",
"ballista-cli",
"ballista/client",
"ballista/core",
"ballista/executor",
"ballista/scheduler",
"benchmarks",
"examples",
"ballista-cli",
]
exclude = ["python"]

Expand Down
4 changes: 2 additions & 2 deletions ballista-cli/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,8 @@ ballista = { path = "../ballista/client", version = "0.10.0", features = [
"standalone",
] }
clap = { version = "3", features = ["derive", "cargo"] }
datafusion = "15.0.0"
datafusion-cli = "15.0.0"
datafusion = "16.1.0"
datafusion-cli = "16.1.0"
dirs = "4.0.0"
env_logger = "0.10"
mimalloc = { version = "0.1", default-features = false }
Expand Down
6 changes: 3 additions & 3 deletions ballista/client/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,12 +31,12 @@ rust-version = "1.63"
ballista-core = { path = "../core", version = "0.10.0" }
ballista-executor = { path = "../executor", version = "0.10.0", optional = true }
ballista-scheduler = { path = "../scheduler", version = "0.10.0", optional = true }
datafusion = "15.0.0"
datafusion-proto = "15.0.0"
datafusion = "16.1.0"
datafusion-proto = "16.1.0"
futures = "0.3"
log = "0.4"
parking_lot = "0.12"
sqlparser = "0.27"
sqlparser = "0.30.0"
tempfile = "3"
tokio = "1.0"

Expand Down
46 changes: 27 additions & 19 deletions ballista/client/src/context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,8 +139,8 @@ impl BallistaContext {
config: &BallistaConfig,
concurrent_tasks: usize,
) -> ballista_core::error::Result<Self> {
use ballista_core::serde::protobuf::PhysicalPlanNode;
use ballista_core::serde::BallistaCodec;
use datafusion_proto::protobuf::PhysicalPlanNode;

log::info!("Running in local mode. Scheduler will be run in-proc");

Expand Down Expand Up @@ -212,7 +212,7 @@ impl BallistaContext {
&self,
path: &str,
options: AvroReadOptions<'_>,
) -> Result<Arc<DataFrame>> {
) -> Result<DataFrame> {
let df = self.context.read_avro(path, options).await?;
Ok(df)
}
Expand All @@ -223,7 +223,7 @@ impl BallistaContext {
&self,
path: &str,
options: ParquetReadOptions<'_>,
) -> Result<Arc<DataFrame>> {
) -> Result<DataFrame> {
let df = self.context.read_parquet(path, options).await?;
Ok(df)
}
Expand All @@ -234,7 +234,7 @@ impl BallistaContext {
&self,
path: &str,
options: CsvReadOptions<'_>,
) -> Result<Arc<DataFrame>> {
) -> Result<DataFrame> {
let df = self.context.read_csv(path, options).await?;
Ok(df)
}
Expand Down Expand Up @@ -262,7 +262,7 @@ impl BallistaContext {
.map_err(|e| {
DataFusionError::Context(format!("Can't read CSV: {}", path), Box::new(e))
})?
.to_logical_plan()?;
.into_optimized_plan()?;
match plan {
LogicalPlan::TableScan(TableScan { source, .. }) => {
self.register_table(name, source_as_provider(&source)?)
Expand All @@ -277,7 +277,11 @@ impl BallistaContext {
path: &str,
options: ParquetReadOptions<'_>,
) -> Result<()> {
match self.read_parquet(path, options).await?.to_logical_plan()? {
match self
.read_parquet(path, options)
.await?
.into_optimized_plan()?
{
LogicalPlan::TableScan(TableScan { source, .. }) => {
self.register_table(name, source_as_provider(&source)?)
}
Expand All @@ -291,7 +295,7 @@ impl BallistaContext {
path: &str,
options: AvroReadOptions<'_>,
) -> Result<()> {
match self.read_avro(path, options).await?.to_logical_plan()? {
match self.read_avro(path, options).await?.into_optimized_plan()? {
LogicalPlan::TableScan(TableScan { source, .. }) => {
self.register_table(name, source_as_provider(&source)?)
}
Expand Down Expand Up @@ -332,7 +336,7 @@ impl BallistaContext {
///
/// This method is `async` because queries of type `CREATE EXTERNAL TABLE`
/// might require the schema to be inferred.
pub async fn sql(&self, sql: &str) -> Result<Arc<DataFrame>> {
pub async fn sql(&self, sql: &str) -> Result<DataFrame> {
let mut ctx = self.context.clone();

let is_show = self.is_show_statement(sql).await?;
Expand Down Expand Up @@ -361,7 +365,7 @@ impl BallistaContext {
}
}

let plan = ctx.create_logical_plan(sql)?;
let plan = ctx.state().create_logical_plan(sql).await?;

match plan {
LogicalPlan::CreateExternalTable(CreateExternalTable {
Expand All @@ -375,7 +379,7 @@ impl BallistaContext {
ref if_not_exists,
..
}) => {
let table_exists = ctx.table_exist(name.as_str())?;
let table_exists = ctx.table_exist(name.as_table_reference())?;
let schema: SchemaRef = Arc::new(schema.as_ref().to_owned().into());
let table_partition_cols = table_partition_cols
.iter()
Expand All @@ -397,37 +401,40 @@ impl BallistaContext {
if !schema.fields().is_empty() {
options = options.schema(&schema);
}
self.register_csv(name, location, options).await?;
Ok(Arc::new(DataFrame::new(ctx.state.clone(), &plan)))
self.register_csv(
name.as_table_reference().table(),
location,
options,
)
.await?;
Ok(DataFrame::new(ctx.state(), plan))
}
"parquet" => {
self.register_parquet(
name,
name.as_table_reference().table(),
location,
ParquetReadOptions::default()
.table_partition_cols(table_partition_cols),
)
.await?;
Ok(Arc::new(DataFrame::new(ctx.state.clone(), &plan)))
Ok(DataFrame::new(ctx.state(), plan))
}
"avro" => {
self.register_avro(
name,
name.as_table_reference().table(),
location,
AvroReadOptions::default()
.table_partition_cols(table_partition_cols),
)
.await?;
Ok(Arc::new(DataFrame::new(ctx.state.clone(), &plan)))
Ok(DataFrame::new(ctx.state(), plan))
}
_ => Err(DataFusionError::NotImplemented(format!(
"Unsupported file type {:?}.",
file_type
))),
},
(true, true) => {
Ok(Arc::new(DataFrame::new(ctx.state.clone(), &plan)))
}
(true, true) => Ok(DataFrame::new(ctx.state(), plan)),
(false, true) => Err(DataFusionError::Execution(format!(
"Table '{:?}' already exists",
name
Expand Down Expand Up @@ -593,6 +600,7 @@ mod tests {
collect_stat: x.collect_stat,
target_partitions: x.target_partitions,
file_sort_order: None,
infinite_source: false,
};

let table_paths = listing_table
Expand Down
8 changes: 4 additions & 4 deletions ballista/core/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,13 +46,13 @@ simd = ["datafusion/simd"]
[dependencies]
ahash = { version = "0.8", default-features = false }

arrow-flight = { version = "28.0.0", features = ["flight-sql-experimental"] }
arrow-flight = { version = "29.0.0", features = ["flight-sql-experimental"] }
async-trait = "0.1.41"
chrono = { version = "0.4", default-features = false }
clap = { version = "3", features = ["derive", "cargo"] }
datafusion = "15.0.0"
datafusion = "16.1.0"
datafusion-objectstore-hdfs = { version = "0.1.1", default-features = false, optional = true }
datafusion-proto = "15.0.0"
datafusion-proto = "16.1.0"
futures = "0.3"
hashbrown = "0.13"

Expand All @@ -68,7 +68,7 @@ prost = "0.11"
prost-types = "0.11"
rand = "0.8"
serde = { version = "1", features = ["derive"] }
sqlparser = "0.27"
sqlparser = "0.30.0"
sys-info = "0.9.0"
tokio = "1.0"
tokio-stream = { version = "0.1", features = ["net"] }
Expand Down
4 changes: 2 additions & 2 deletions ballista/core/build.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ fn main() -> Result<(), String> {
println!("cargo:rerun-if-env-changed=FORCE_REBUILD");

let version = rustc_version::version().unwrap();
println!("cargo:rustc-env=RUSTC_VERSION={}", version);
println!("cargo:rustc-env=RUSTC_VERSION={version}");

// TODO: undo when resolved: https://github.com/intellij-rust/intellij-rust/issues/9402
#[cfg(feature = "docsrs")]
Expand All @@ -42,7 +42,7 @@ fn main() -> Result<(), String> {
tonic_build::configure()
.extern_path(".datafusion", "::datafusion_proto::protobuf")
.compile(&["proto/ballista.proto"], &["proto"])
.map_err(|e| format!("protobuf compilation failed: {}", e))?;
.map_err(|e| format!("protobuf compilation failed: {e}"))?;
let generated_source_path = out.join("ballista.protobuf.rs");
let code = std::fs::read_to_string(generated_source_path).unwrap();
let mut file = std::fs::OpenOptions::new()
Expand Down
Loading