Skip to content

feat: upgrade to DataFusion 47.0.0 #3378

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

Merged
merged 4 commits into from
May 1, 2025
Merged

feat: upgrade to DataFusion 47.0.0 #3378

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1e26ebc
feat: Update to Datafusion 47.0.0
alamb Apr 11, 2025
d664b37
feat: Update to Datafusion 47.0.0
alamb Apr 11, 2025
1166679
chore: re-enable hdfs support and add a teensy tiny unit test
rtyler May 1, 2025
837ead8
chore: tighten up the checking on the predicate comparison
rtyler May 1, 2025
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
48 changes: 24 additions & 24 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,36 +27,36 @@ debug = "line-tables-only"

[workspace.dependencies]
delta_kernel = { version = "0.10.0", features = [
"arrow-54",
"arrow-55",
"internal-api",
] }

# arrow
arrow = { version = "54" }
arrow-arith = { version = "54" }
arrow-array = { version = "54", features = ["chrono-tz"] }
arrow-buffer = { version = "54" }
arrow-cast = { version = "54" }
arrow-ipc = { version = "54" }
arrow-json = { version = "54" }
arrow-ord = { version = "54" }
arrow-row = { version = "54" }
arrow-schema = { version = "54" }
arrow-select = { version = "54" }
object_store = { version = "0.11.2" }
parquet = { version = "54" }
arrow = { version = "55" }
arrow-arith = { version = "55" }
arrow-array = { version = "55", features = ["chrono-tz"] }
arrow-buffer = { version = "55" }
arrow-cast = { version = "55" }
arrow-ipc = { version = "55" }
arrow-json = { version = "55" }
arrow-ord = { version = "55" }
arrow-row = { version = "55" }
arrow-schema = { version = "55" }
arrow-select = { version = "55" }
object_store = { version = "0.12.0" }
parquet = { version = "55" }

# datafusion
datafusion = "46"
datafusion-expr = "46"
datafusion-common = "46"
datafusion-ffi = "46"
datafusion-functions = "46"
datafusion-functions-aggregate = "46"
datafusion-physical-expr = "46"
datafusion-physical-plan = "46"
datafusion-proto = "46"
datafusion-sql = "46"
datafusion = "47.0.0"
datafusion-expr = "47.0.0"
datafusion-common = "47.0.0"
datafusion-ffi = "47.0.0"
datafusion-functions = "47.0.0"
datafusion-functions-aggregate = "47.0.0"
datafusion-physical-expr = "47.0.0"
datafusion-physical-plan = "47.0.0"
datafusion-proto = "47.0.0"
datafusion-sql = "47.0.0"

# serde
serde = { version = "1.0.194", features = ["derive"] }
Expand Down
6 changes: 3 additions & 3 deletions crates/aws/src/storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -358,7 +358,7 @@ impl ObjectStore for S3StorageBackend {
self.inner.get_opts(location, options).await
}

async fn get_range(&self, location: &Path, range: Range<usize>) -> ObjectStoreResult<Bytes> {
async fn get_range(&self, location: &Path, range: Range<u64>) -> ObjectStoreResult<Bytes> {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self.inner.get_range(location, range).await
}

Expand All @@ -370,15 +370,15 @@ impl ObjectStore for S3StorageBackend {
self.inner.delete(location).await
}

fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, ObjectStoreResult<ObjectMeta>> {
fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self.inner.list(prefix)
}

fn list_with_offset(
&self,
prefix: Option<&Path>,
offset: &Path,
) -> BoxStream<'_, ObjectStoreResult<ObjectMeta>> {
) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
self.inner.list_with_offset(prefix, offset)
}

Expand Down
6 changes: 3 additions & 3 deletions crates/aws/tests/repair_s3_rename_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,7 @@ impl ObjectStore for DelayedObjectStore {
self.inner.get_opts(location, options).await
}

async fn get_range(&self, location: &Path, range: Range<usize>) -> ObjectStoreResult<Bytes> {
async fn get_range(&self, location: &Path, range: Range<u64>) -> ObjectStoreResult<Bytes> {
self.inner.get_range(location, range).await
}

Expand All @@ -199,15 +199,15 @@ impl ObjectStore for DelayedObjectStore {
self.inner.delete(location).await
}

fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, ObjectStoreResult<ObjectMeta>> {
fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
self.inner.list(prefix)
}

fn list_with_offset(
&self,
prefix: Option<&Path>,
offset: &Path,
) -> BoxStream<'_, ObjectStoreResult<ObjectMeta>> {
) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
self.inner.list_with_offset(prefix, offset)
}

Expand Down
3 changes: 2 additions & 1 deletion crates/azure/tests/integration.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,8 @@ async fn read_write_test_onelake(context: &IntegrationContext, path: &Path) -> T
assert_eq!(expected, fetched);

for range in [0..10, 3..5, 0..expected.len()] {
let data = delta_store.get_range(path, range.clone()).await.unwrap();
let range_u64 = range.start as u64..range.end as u64;
let data = delta_store.get_range(path, range_u64).await.unwrap();
assert_eq!(&data[..], &expected[range])
}

Expand Down
2 changes: 1 addition & 1 deletion crates/core/src/delta_datafusion/cdf/scan_utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ pub fn create_partition_values<F: FileAction>(
let part = PartitionedFile {
object_meta: ObjectMeta {
location: Path::parse(action.path().as_str())?,
size: action.size()?,
size: action.size()? as u64,
e_tag: None,
last_modified: chrono::Utc.timestamp_nanos(0),
version: None,
Expand Down
9 changes: 0 additions & 9 deletions crates/core/src/delta_datafusion/logical.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,15 +50,6 @@ impl UserDefinedLogicalNodeCore for MetricObserver {
write!(f, "MetricObserver id={}", self.id)
}

fn from_template(
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is an old api that was removed

&self,
exprs: &[datafusion_expr::Expr],
inputs: &[datafusion_expr::LogicalPlan],
) -> Self {
self.with_exprs_and_inputs(exprs.to_vec(), inputs.to_vec())
.unwrap()
}

fn with_exprs_and_inputs(
&self,
_exprs: Vec<datafusion_expr::Expr>,
Expand Down
38 changes: 19 additions & 19 deletions crates/core/src/delta_datafusion/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,12 @@ use arrow_schema::{
use arrow_select::concat::concat_batches;
use async_trait::async_trait;
use chrono::{DateTime, TimeZone, Utc};
use datafusion::catalog::memory::DataSourceExec;
use datafusion::catalog::{Session, TableProviderFactory};
use datafusion::config::TableParquetOptions;
use datafusion::datasource::physical_plan::{
wrap_partition_type_in_dict, wrap_partition_value_in_dict, FileScanConfig, ParquetSource,
wrap_partition_type_in_dict, wrap_partition_value_in_dict, FileGroup, FileScanConfigBuilder,
ParquetSource,
};
use datafusion::datasource::{listing::PartitionedFile, MemTable, TableProvider, TableType};
use datafusion::execution::context::{SessionConfig, SessionContext, SessionState, TaskContext};
Expand Down Expand Up @@ -671,7 +673,7 @@ impl<'a> DeltaScanBuilder<'a> {
}
};

let file_scan_config = FileScanConfig::new(
let file_scan_config = FileScanConfigBuilder::new(
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

self.log_store.object_store_url(),
file_schema,
Arc::new(file_source),
Expand All @@ -682,15 +684,16 @@ impl<'a> DeltaScanBuilder<'a> {
//
// See https://github.com/apache/datafusion/issues/11322
if file_groups.is_empty() {
vec![vec![]]
vec![FileGroup::from(vec![])]
} else {
file_groups.into_values().collect()
file_groups.into_values().map(FileGroup::from).collect()
},
)
.with_statistics(stats)
.with_projection(self.projection.cloned())
.with_limit(self.limit)
.with_table_partition_cols(table_partition_cols);
.with_table_partition_cols(table_partition_cols)
.build();

let metrics = ExecutionPlanMetricsSet::new();
MetricBuilder::new(&metrics)
Expand All @@ -702,7 +705,7 @@ impl<'a> DeltaScanBuilder<'a> {

Ok(DeltaScan {
table_uri: ensure_table_uri(self.log_store.root_uri())?.as_str().into(),
parquet_scan: file_scan_config.build(),
parquet_scan: DataSourceExec::from_data_source(file_scan_config),
config,
logical_schema,
metrics,
Expand Down Expand Up @@ -1974,6 +1977,7 @@ mod tests {
use bytes::Bytes;
use chrono::{TimeZone, Utc};
use datafusion::assert_batches_sorted_eq;
use datafusion::datasource::physical_plan::FileScanConfig;
use datafusion::datasource::source::DataSourceExec;
use datafusion::physical_plan::empty::EmptyExec;
use datafusion::physical_plan::{visit_execution_plan, ExecutionPlanVisitor, PhysicalExpr};
Expand Down Expand Up @@ -2731,10 +2735,6 @@ mod tests {
visit_execution_plan(&scan, &mut visitor).unwrap();

assert_eq!(visitor.predicate.unwrap().to_string(), "a@0 = s");
assert_eq!(
visitor.pruning_predicate.unwrap().orig_expr().to_string(),
"a@0 = s"
);
}

#[tokio::test]
Expand Down Expand Up @@ -2766,7 +2766,6 @@ mod tests {
visit_execution_plan(&scan, &mut visitor).unwrap();

assert!(visitor.predicate.is_none());
assert!(visitor.pruning_predicate.is_none());
}

#[tokio::test]
Expand Down Expand Up @@ -2801,7 +2800,6 @@ mod tests {
#[derive(Default)]
struct ParquetVisitor {
predicate: Option<Arc<dyn PhysicalExpr>>,
pruning_predicate: Option<Arc<PruningPredicate>>,
options: Option<TableParquetOptions>,
}

Expand All @@ -2828,7 +2826,6 @@ mod tests {
{
self.options = Some(parquet_source.table_parquet_options().clone());
self.predicate = parquet_source.predicate().cloned();
self.pruning_predicate = parquet_source.pruning_predicate().cloned();
}

Ok(true)
Expand Down Expand Up @@ -2974,8 +2971,8 @@ mod tests {

#[derive(Debug, PartialEq)]
enum ObjectStoreOperation {
GetRanges(LocationType, Vec<Range<usize>>),
GetRange(LocationType, Range<usize>),
GetRanges(LocationType, Vec<Range<u64>>),
GetRange(LocationType, Range<u64>),
GetOpts(LocationType),
Get(LocationType),
}
Expand Down Expand Up @@ -3054,7 +3051,7 @@ mod tests {
async fn get_range(
&self,
location: &Path,
range: Range<usize>,
range: Range<u64>,
) -> object_store::Result<Bytes> {
self.operations
.send(ObjectStoreOperation::GetRange(
Expand All @@ -3068,7 +3065,7 @@ mod tests {
async fn get_ranges(
&self,
location: &Path,
ranges: &[Range<usize>],
ranges: &[Range<u64>],
) -> object_store::Result<Vec<Bytes>> {
self.operations
.send(ObjectStoreOperation::GetRanges(
Expand All @@ -3094,15 +3091,18 @@ mod tests {
self.inner.delete_stream(locations)
}

fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, object_store::Result<ObjectMeta>> {
fn list(
&self,
prefix: Option<&Path>,
) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
self.inner.list(prefix)
}

fn list_with_offset(
&self,
prefix: Option<&Path>,
offset: &Path,
) -> BoxStream<'_, object_store::Result<ObjectMeta>> {
) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
self.inner.list_with_offset(prefix, offset)
}

Expand Down
19 changes: 0 additions & 19 deletions crates/core/src/delta_datafusion/schema_adapter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,6 @@ impl SchemaAdapter for DeltaSchemaAdapter {
Ok((
Arc::new(SchemaMapping {
projected_schema: self.projected_table_schema.clone(),
table_schema: self.table_schema.clone(),
}),
projection,
))
Expand All @@ -69,29 +68,11 @@ impl SchemaAdapter for DeltaSchemaAdapter {
#[derive(Debug)]
pub(crate) struct SchemaMapping {
projected_schema: SchemaRef,
table_schema: SchemaRef,
}

impl SchemaMapper for SchemaMapping {
fn map_batch(&self, batch: RecordBatch) -> datafusion_common::Result<RecordBatch> {
let record_batch = cast_record_batch(&batch, self.projected_schema.clone(), false, true)?;
Ok(record_batch)
}

fn map_partial_batch(&self, batch: RecordBatch) -> datafusion_common::Result<RecordBatch> {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Due to @adriangb simplifying the implementation in

let partial_table_schema = Arc::new(Schema::new(
batch
.schema()
.fields()
.iter()
.filter_map(|batch_field| {
self.table_schema.field_with_name(batch_field.name()).ok()
})
.cloned()
.collect::<Vec<_>>(),
));

let record_batch = cast_record_batch(&batch, partial_table_schema, false, true)?;
Ok(record_batch)
}
}
2 changes: 1 addition & 1 deletion crates/core/src/kernel/snapshot/log_data.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -391,7 +391,7 @@ impl<'a> TryFrom<&LogicalFile<'a>> for ObjectMeta {
fn try_from(file_stats: &LogicalFile<'a>) -> Result<Self, Self::Error> {
Ok(ObjectMeta {
location: file_stats.object_store_path(),
size: file_stats.size() as usize,
size: file_stats.size() as u64,
last_modified: file_stats.modification_datetime()?,
version: None,
e_tag: None,
Expand Down
7 changes: 4 additions & 3 deletions crates/core/src/kernel/snapshot/log_segment.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -300,7 +300,8 @@ impl LogSegment {
let store = store.clone();
let read_schema = read_schema.clone();
async move {
let mut reader = ParquetObjectReader::new(store, meta);
let mut reader =
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ParquetObjectReader::new(store, meta.location).with_file_size(meta.size);
let options = ArrowReaderOptions::new();
let reader_meta = ArrowReaderMetadata::load_async(&mut reader, options).await?;

Expand Down Expand Up @@ -413,7 +414,7 @@ impl LogSegment {
let bytes = commit.get_bytes()?;
let meta = ObjectMeta {
location: path,
size: bytes.len(),
size: bytes.len() as u64,
last_modified: Utc::now(),
e_tag: None,
version: None,
Expand Down Expand Up @@ -777,7 +778,7 @@ pub(super) mod tests {
self.store.delete(location).await
}

fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>> {
fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
std::thread::sleep(std::time::Duration::from_secs(1));
self.store.list(prefix)
}
Expand Down
Loading
Loading