GeezerVison is showing github.com — [adapters] Delta: let a filter name a nested column-mapped field by blp · Pull Request #7669 · feldera/feldera
[adapters] Delta: let a filter name a nested column-mapped field
#7669
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.
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
Uh oh!
There was an error while loading. Please reload this page.
Show file tree
Hide file tree
Changes from all commits
Commits
File filter
Filter by extension
.py
(1)
.rs
(2)
All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
Original file line number
Diff line number
Diff line change
Expand Up
@@ -29,6 +29,10 @@ use datafusion::datasource::listing::{
};
use datafusion::execution::memory_pool::MemoryLimit;
use datafusion::logical_expr::physical_planning_context::PhysicalPlanningContext;
use datafusion::logical_expr::{
ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
Volatility,
};
use datafusion::optimizer::analyzer::type_coercion::TypeCoercionRewriter;
use datafusion::physical_plan::{
ExecutionPlan, ExecutionPlanProperties, PhysicalExpr, SendableRecordBatchStream,
Expand Down
Expand Up
@@ -80,6 +84,7 @@ use std::cmp::min;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::fmt::Display;
use std::future::Future;
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::thread;
Expand Down
Expand Up
@@ -544,7 +549,9 @@ fn relabel_data_type(data_type: &ArrowDataType, map: &HashMap<String, String>) -
/// Translate a batch's nested field names physical-to-logical. Top-level names
/// are left as-is (already logical); only names nested inside a struct, list, or
/// map are rewritten.
/// map are rewritten. Only tests use it: queries relabel through
/// [`RelabelNestedUdf`].
#[cfg(test)]
fn relabel_nested_columns(
batch: &RecordBatch,
map: &HashMap<String, String>,
Expand Down
Expand Up
@@ -573,6 +580,57 @@ fn relabel_nested_columns(
.map_err(|e| anyhow!("relabeling column-mapped field names failed: {e}"))
}
/// Scalar function that applies [`relabel_array`] to its one argument, so that
/// a query can name nested fields by their logical names (see
/// [`DeltaTableInputEndpointInner::project_physical_to_logical`]).
#[derive(Debug, PartialEq, Eq)]
struct RelabelNestedUdf {
signature: Signature,
map: HashMap<String, String>,
}
impl Hash for RelabelNestedUdf {
fn hash<H: Hasher>(&self, state: &mut H) {
// `HashMap` is not `Hash`. Equal values still hash equal.
self.name().hash(state);
}
}
impl ScalarUDFImpl for RelabelNestedUdf {
fn name(&self) -> &str {
"feldera_relabel_nested"
}
fn signature(&self) -> &Signature {
&self.signature
}
fn return_type(&self, arg_types: &[ArrowDataType]) -> Result<ArrowDataType, DataFusionError> {
Ok(relabel_data_type(&arg_types[0], &self.map))
}
fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef, DataFusionError> {
let arg = &args.arg_fields[0];
Ok(Arc::new(ArrowField::new(
self.name(),
relabel_data_type(arg.data_type(), &self.map),
arg.is_nullable(),
)))
}
fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue, DataFusionError> {
let relabel = |array: &ArrayRef| {
relabel_array(array, &self.map).map_err(|e| DataFusionError::External(e.into()))
};
match &args.args[0] {
ColumnarValue::Array(array) => Ok(ColumnarValue::Array(relabel(array)?)),
ColumnarValue::Scalar(scalar) => Ok(ColumnarValue::Scalar(
ScalarValue::try_from_array(&relabel(&scalar.to_array()?)?, 0)?,
)),
}
}
}
/// Apply the connector's optional `filter` to `df`, or return `df` unchanged.
/// The expression parses against `df`'s own schema, so `df` must still carry
/// every column the filter names. `relation` names the frame when a caller has
Expand Down
Expand Up
@@ -3725,18 +3783,6 @@ impl DeltaTableInputEndpointInner {
self.consumer
.update_connector_health(ConnectorHealth::healthy());
// Nested struct fields are read under physical names; restore their
// logical names per batch (top-level names are already logical). Empty,
// a no-op, unless the table nests column-mapped fields.
let nested_map = nested_physical_to_logical(
self.schema_snapshot()
.snapshot()
.map_err(|e| format!("error accessing Delta table snapshot: {e}"))?
.snapshot()
.arrow_schema()
.as_ref(),
);
let mut num_batches = 0;
let mut total_records = 0usize;
Expand Down
Expand Up
@@ -3816,11 +3862,6 @@ impl DeltaTableInputEndpointInner {
));
}
};
let batch = if nested_map.is_empty() {
batch
} else {
relabel_nested_columns(&batch, &nested_map).map_err(|e| e.to_string())?
};
num_batches += 1;
total_records += batch.num_rows();
Expand Down
Expand Up
@@ -4760,8 +4801,8 @@ impl DeltaTableInputEndpointInner {
/// the logical schema restricted to columns `keep` accepts (in schema order,
/// so read sides line up), with each column-mapped field renamed to its
/// physical (`col-<uuid>`) name so DataFusion matches by name. It recurses
/// into nested struct, list, and map fields. Logical names are restored afterwards:
/// top level in `project_physical_to_logical`, nested in `relabel_nested_columns `.
/// into nested struct, list, and map fields. Logical names are restored afterwards,
/// at every level, in `project_physical_to_logical `.
/// With column mapping off this is just the kept logical schema.
///
/// Partition columns are always excluded: Delta never stores them in the data
Expand Down
Expand Up
@@ -4791,6 +4832,9 @@ impl DeltaTableInputEndpointInner {
/// (already logical, or Delta metadata like `__feldera_op`) pass through.
/// This function is a no-op when column mapping is off.
///
/// Nested field names are restored too, through [`RelabelNestedUdf`], so
/// that a filter or ordering that follows can name them (#7657).
///
/// Names are taken verbatim, with `ident`, for the reason
/// [`partition_projection`](Self::partition_projection) gives.
fn project_physical_to_logical(&self, df: DataFrame) -> AnyResult<DataFrame> {
Expand All
@@ -4802,13 +4846,28 @@ impl DeltaTableInputEndpointInner {
.iter()
.map(|(l, p)| (p.as_str(), l.as_str()))
.collect();
let nested_map = nested_physical_to_logical(self.logical_schema()?.as_ref());
let has_nested =
|f: &ArrowField| relabel_data_type(f.data_type(), &nested_map) != *f.data_type();
let relabel = ScalarUDF::new_from_impl(RelabelNestedUdf {
signature: Signature::any(1, Volatility::Immutable),
map: nested_map.clone(),
});
let exprs: Vec<Expr> = df
.schema()
.fields()
.iter()
.map(|f| match to_logical.get(f.name().as_str()) {
Some(logical) => ident(f.name()).alias(*logical),
None => ident(f.name()),
.map(|f| {
let logical = to_logical.get(f.name().as_str()).copied();
if has_nested(f) {
relabel
.call(vec![ident(f.name())])
.alias(logical.unwrap_or(f.name()))
} else if let Some(logical) = logical {
ident(f.name()).alias(logical)
} else {
ident(f.name())
}
})
.collect();
df.select(exprs)
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -6016,7 +6016,7 @@ pub fn change_feed_always_transaction_test(engine: &dyn PipelineTestEngine) {
///
/// Read over [`relation`], whose 21 columns carry the types the change
/// data path handles with machinery of its own: `change_data_read_schema`,
/// `project_physical_to_logical` with `relabel_nested_columns `, and
/// `project_physical_to_logical` with `RelabelNestedUdf `, and
/// `add_partition_columns`. Binary, decimal, arrays, nested rows, maps of rows,
/// `VARIANT` and `UUID` are where that machinery breaks, and none of them is
/// reached by four scalar columns. The oracle is the follow-mode connector
Expand Down
Expand Up
@@ -8270,6 +8270,147 @@ pub fn follow_id_mapped_filter_test(engine: &dyn PipelineTestEngine) {
})
}
/// The rows of a column-mapped table with three distinct nested `after.status`
/// values, for the tests that filter on a nested field.
fn nested_filter_rows() -> [Value; 3] {
[
uniform_row(1, "alpha", "Coffee Shop", "settled"),
uniform_row(2, "beta", "Gas Station", "pending"),
uniform_row(3, "gamma", "Book Store", "reversed"),
]
}
/// `filter` can name a field nested in a column-mapped struct when following
/// the table. The data file names the nested field `col-<id>`, so the filter
/// resolves only if the read restores logical names before it filters (#7657).
#[engine_test]
pub fn follow_id_mapped_nested_filter_test(engine: &dyn PipelineTestEngine) {
run_async_on(2, async {
let table_dir = TempDir::new().unwrap();
let rows = nested_filter_rows();
write_id_mapped_table_physical_names(table_dir.path(), &uniform_relation(), &[&rows]);
let table_uri = table_dir.path().display().to_string();
let read = off_runtime(move || {
read_uniform_table(
engine,
&table_uri,
1,
&[
("mode", json!("follow")),
("filter", json!("after.status <> 'pending'")),
],
)
});
assert_eq!(
read,
rows_zset(&uniform_relation(), &[rows[0].clone(), rows[2].clone()]),
"the nested filter must drop row 2 and keep the rest"
);
})
}
/// `filter`, `cdc_delete_filter`, and `cdc_order_by` can name fields nested in
/// a column-mapped struct in CDC mode. CDC reads files through the same
/// `files_dataframe` as the snapshot of a distributed connector (#7657).
#[engine_test]
pub fn cdc_id_mapped_nested_filter_test(engine: &dyn PipelineTestEngine) {
run_async_on(2, async {
let table_dir = TempDir::new().unwrap();
let rows = nested_filter_rows();
write_id_mapped_table_physical_names(table_dir.path(), &uniform_relation(), &[&rows]);
let table_uri = table_dir.path().display().to_string();
let read = off_runtime(move || {
read_uniform_table(
engine,
&table_uri,
1,
&[
("mode", json!("cdc")),
("filter", json!("after.merchant <> 'Coffee Shop'")),
// Matches no row: the table has no key, so a delete of a row
// never inserted would count -1.
("cdc_delete_filter", json!("after.status = 'none'")),
("cdc_order_by", json!("after.merchant")),
],
)
});
assert_eq!(
read,
rows_zset(&uniform_relation(), &[rows[1].clone(), rows[2].clone()]),
"the nested filter must drop row 1 and keep the rest"
);
})
}
/// `filter` can name a field nested in a column-mapped struct in a commit read
/// from its change data file (#7657).
#[engine_test]
pub fn follow_id_mapped_change_data_nested_filter_test(engine: &dyn PipelineTestEngine) {
run_async_on(2, async {
let table_dir = TempDir::new().unwrap();
let rows = [
uniform_cdc_row(1, "alpha", "Coffee Shop", "settled", "c", 1),
uniform_cdc_row(2, "beta", "Gas Station", "pending", "c", 2),
];
// The change data commit inserts a settled row, a pending row, and
// deletes row 1.
let changes = [
(
uniform_cdc_row(3, "gamma", "Book Store", "settled", "c", 3),
"insert",
),
(
uniform_cdc_row(4, "delta", "Bakery", "pending", "c", 4),
"insert",
),
(
uniform_cdc_row(1, "alpha", "Coffee Shop", "settled", "d", 5),
"delete",
),
];
write_id_mapped_change_data_table(
table_dir.path(),
&uniform_cdc_relation(),
&rows,
&changes,
true,
);
let table_uri = table_dir.path().display().to_string();
let (read, change_data_commits) = off_runtime(move || {
let mut commits = 0;
let read = read_id_mapped_table_counting(
engine,
&table_uri,
&uniform_relation(),
2,
&[
("mode", json!("follow")),
("change_feed", json!("require")),
("filter", json!("after.status = 'settled'")),
],
&mut commits,
);
(read, commits)
});
assert_eq!(change_data_commits, 1);
assert_eq!(
read,
rows_zset(
&uniform_relation(),
&[uniform_row(3, "gamma", "Book Store", "settled")]
),
"the nested filter must keep only settled rows, and the delete must \
cancel row 1"
);
})
}
/// Every SQL type the connector supports survives a `mode = 'id'` follow read.
///
/// `mode = 'id'` routes every file through `project_to_logical` instead of
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -29,14 +29,19 @@
_FIXTURE_BUILDER = Path(__file__).parent / "fixtures" / "column_mapping_nested.py"
def _build_sql(loc: DeltaTestLocation) -> str:
def _build_sql(
loc: DeltaTestLocation,
extra_config: dict[str, object] | None = None,
distributed: bool = False,
) -> str:
config = dict(loc.connector_config)
# Replay v1 + v2 (v0 is the empty CREATE) through the follow path.
config.update({"version": 0, "end_version": 2})
config.update(extra_config or {"version": 0, "end_version": 2})
connectors = json.dumps(
[
{
"name": CONNECTOR,
"distributed": distributed,
"transport": {"name": "delta_table_input", "config": config},
}
]
Expand Down
Expand Up
@@ -103,3 +108,43 @@ def test_delta_input_column_mapping_nested_follow(pipeline_name):
pipeline.stop(force=True)
finally:
loc.cleanup()
def test_delta_input_column_mapping_nested_dist_filter(pipeline_name):
"""A distributed snapshot of a nested column-mapped table applies a
``filter`` that names a nested field (#7657). Each host reads its part of
the snapshot as raw Parquet, so the nested names must be logical before the
filter runs."""
loc = DeltaTestLocation.create(
pipeline_name,
mode="snapshot",
stable_subpath=f"column_mapping_nested_{FIXTURE_VERSION}",
)
try:
ensure_delta_spark_fixture(loc, _FIXTURE_BUILDER)
pipeline = PipelineBuilder(
TEST_CLIENT,
pipeline_name,
sql=_build_sql(
loc,
{"filter": "after.transaction__amount <> '20'"},
distributed=True,
),
runtime_config=RuntimeConfig(
workers=FELDERA_TEST_NUM_WORKERS,
hosts=FELDERA_TEST_NUM_HOSTS,
),
).create_or_replace()
pipeline.start()
pipeline.wait_for_completion(force_stop=False, timeout_s=600)
ids = sorted(
r["id"]
for r in pipeline.query(f"SELECT after.transaction__id AS id FROM {TABLE}")
)
assert ids == ["t1", "t3"], f"the nested filter must drop t2; got {ids}"
pipeline.stop(force=True)
finally:
loc.cleanup()
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.