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

feat(python, rust): arrow large/view types passthrough, rust default engine #2738

Merged
merged 5 commits into from
Aug 10, 2024
Merged

feat(python, rust): arrow large/view types passthrough, rust default engine #2738

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9f336d1
feat: arrow large/view type passthrough + refactor writer
ion-elgreco Aug 9, 2024
913098a
feat: pyarrow view/large types passthrough in python
ion-elgreco Aug 9, 2024
0daaf03
feat: change default to rust writer engine
ion-elgreco Aug 9, 2024
002fbe4
chore: bump python 0.19
ion-elgreco Aug 9, 2024
a617e48
chore: improve docs + CI
ion-elgreco Aug 10, 2024
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
10 changes: 2 additions & 8 deletions .github/workflows/python_build.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ jobs:
run: make check-rust

test-minimal:
name: Python Build (Python 3.8 PyArrow 8.0.0)
name: Python Build (Python 3.8 PyArrow latest)
runs-on: ubuntu-latest
env:
RUSTFLAGS: "-C debuginfo=line-tables-only"
Expand All @@ -49,9 +49,7 @@ jobs:
run: |
python -m venv venv
source venv/bin/activate
make setup
# Install minimum PyArrow version
pip install -e .[pandas,devel] pyarrow==8.0.0
make develop
env:
RUSTFLAGS: "-C debuginfo=line-tables-only"

Expand All @@ -60,10 +58,6 @@ jobs:
source venv/bin/activate
make unit-test

# - name: Run Integration tests
# run: |
# py.test --cov tests -m integration

test:
name: Python Build (Python 3.10 PyArrow latest)
runs-on: ubuntu-latest
Expand Down
2 changes: 1 addition & 1 deletion crates/core/src/delta_datafusion/find_files/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ async fn scan_table_by_files(
.with_file_column(true)
.build(&snapshot)?;

let logical_schema = df_logical_schema(&snapshot, &scan_config)?;
let logical_schema = df_logical_schema(&snapshot, &scan_config.file_column_name, None)?;

// Identify which columns we need to project
let mut used_columns = expression
Expand Down
22 changes: 15 additions & 7 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 @@ -305,9 +305,13 @@ pub(crate) fn register_store(store: LogStoreRef, env: Arc<RuntimeEnv>) {
/// at the physical level
pub(crate) fn df_logical_schema(
snapshot: &DeltaTableState,
scan_config: &DeltaScanConfig,
file_column_name: &Option<String>,
schema: Option<ArrowSchemaRef>,
) -> DeltaResult<SchemaRef> {
let input_schema = snapshot.arrow_schema()?;
let input_schema = match schema {
Some(schema) => schema,
None => snapshot.input_schema()?,
};
Comment on lines +311 to +314
Copy link
Collaborator

Choose a reason for hiding this comment

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

this is interesting, since there seem to be some issues with the wrapped schema also in the pruning logic, which I have yet to fully grok. Why did we need to change here?

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

If we take the arrow_schema it was always wrapping partition values in dictionary, somewhere down the line where it was going to do divide by partition values, it would fail because a slice to get a value from a DictionaryArray will always return None

let table_partition_cols = &snapshot.metadata().partition_columns;

let mut fields: Vec<Arc<Field>> = input_schema
Expand All @@ -326,7 +330,7 @@ pub(crate) fn df_logical_schema(
));
}

if let Some(file_column_name) = &scan_config.file_column_name {
if let Some(file_column_name) = file_column_name {
fields.push(Arc::new(Field::new(file_column_name, DataType::Utf8, true)));
}

Expand Down Expand Up @@ -528,7 +532,11 @@ impl<'a> DeltaScanBuilder<'a> {
None => self.snapshot.arrow_schema(),
}?;

let logical_schema = df_logical_schema(self.snapshot, &config)?;
let logical_schema = df_logical_schema(
self.snapshot,
&config.file_column_name,
Some(schema.clone()),
)?;

let logical_schema = if let Some(used_columns) = self.projection {
let mut fields = vec![];
Expand Down Expand Up @@ -733,7 +741,7 @@ impl TableProvider for DeltaTable {
filter: &[&Expr],
) -> DataFusionResult<Vec<TableProviderFilterPushDown>> {
Ok(filter
.into_iter()
.iter()
.map(|_| TableProviderFilterPushDown::Inexact)
.collect())
}
Expand All @@ -760,7 +768,7 @@ impl DeltaTableProvider {
config: DeltaScanConfig,
) -> DeltaResult<Self> {
Ok(DeltaTableProvider {
schema: df_logical_schema(&snapshot, &config)?,
schema: df_logical_schema(&snapshot, &config.file_column_name, config.schema.clone())?,
snapshot,
log_store,
config,
Expand Down Expand Up @@ -1524,7 +1532,7 @@ pub(crate) async fn find_files_scan<'a>(
}
.build(snapshot)?;

let logical_schema = df_logical_schema(snapshot, &scan_config)?;
let logical_schema = df_logical_schema(snapshot, &scan_config.file_column_name, None)?;

// Identify which columns we need to project
let mut used_columns = expression
Expand Down
5 changes: 2 additions & 3 deletions crates/core/src/operations/add_column.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,10 @@ use delta_kernel::schema::StructType;
use futures::future::BoxFuture;
use itertools::Itertools;

use super::cast::merge_struct;
use super::transaction::{CommitBuilder, CommitProperties, PROTOCOL};

use crate::kernel::StructField;
use crate::logstore::LogStoreRef;
use crate::operations::cast::merge_schema::merge_delta_struct;
use crate::protocol::DeltaOperation;
use crate::table::state::DeltaTableState;
use crate::{DeltaResult, DeltaTable, DeltaTableError};
Expand Down Expand Up @@ -67,7 +66,7 @@ impl std::future::IntoFuture for AddColumnBuilder {

let fields_right = &StructType::new(fields.clone());
let table_schema = this.snapshot.schema();
let new_table_schema = merge_struct(table_schema, fields_right)?;
let new_table_schema = merge_delta_struct(table_schema, fields_right)?;

// TODO(ion): Think of a way how we can simply this checking through the API or centralize some checks.
let contains_timestampntz = PROTOCOL.contains_timestampntz(fields.iter());
Expand Down
Loading
Loading