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

fix(udf): eval_row panic due to misuse of input row as arg row #9240

Merged
merged 2 commits into from
Apr 17, 2023
Merged

fix(udf): eval_row panic due to misuse of input row as arg row #9240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5ebc868
fix(udf): eval_row input row to chunk adapter
xiangjinwu Apr 17, 2023
c047556
tests
xiangjinwu Apr 17, 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
28 changes: 28 additions & 0 deletions e2e_test/udf/python.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,34 @@ select (extract_tcp_info(E'\\x45000034a8a8400040065b8ac0a8000ec0a80001035d20b6d9
----
192.168.0.14 192.168.0.1 861 8374

# error handling

statement error
select hex_to_dec('1z');

statement ok
create table t (dummy date, v varchar);

statement ok
create materialized view mv as select dummy, hex_to_dec(v) from t;

statement ok
insert into t values ('2023-01-01', '1z');
Comment on lines +134 to +138
Copy link
Contributor Author

Choose a reason for hiding this comment

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

With the dummy column, it triggers panic:

thread 'risingwave-streaming-actor' panicked at 'Failed to append datum, array builder type: Utf8, scalar type: Date', src/common/src/array/mod.rs:562:1

Without the dummy column, it triggers panic:

thread 'risingwave-streaming-actor' panicked at 'itertools: .zip_eq() reached end of one iterator before the other', /Users/xiangjin/.cargo/registry/src/index.crates.io-6f17d22bba15001f/itertools-0.10.5/src/zip_eq_impl.rs:48:13


statement ok
flush;

query TI
select * from mv;
----
2023-01-01 NULL

statement ok
drop materialized view mv;

statement ok
drop table t;

# TODO: drop function without arguments

# # Drop a function but ambiguous.
Expand Down
40 changes: 30 additions & 10 deletions src/expr/src/expr/expr_udf.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,36 @@ impl Expression for UdfExpression {
let array = child.eval_checked(input).await?;
columns.push(array.as_ref().into());
}
let opts =
arrow_array::RecordBatchOptions::default().with_row_count(Some(input.capacity()));
self.eval_inner(columns, vis).await
}

async fn eval_row(&self, input: &OwnedRow) -> Result<Datum> {
let mut columns = Vec::with_capacity(self.children.len());
for child in &self.children {
let datum = child.eval_row(input).await?;
columns.push(datum);
}
let arg_row = OwnedRow::new(columns);
let chunk = DataChunk::from_rows(std::slice::from_ref(&arg_row), &self.arg_types);
let arg_columns = chunk
.columns()
.iter()
.map(|c| c.array_ref().into())
.collect();
let output_array = self
.eval_inner(arg_columns, chunk.vis().to_bitmap())
.await?;
Ok(output_array.to_datum())
}
}

impl UdfExpression {
async fn eval_inner(
&self,
columns: Vec<arrow_array::ArrayRef>,
vis: risingwave_common::buffer::Bitmap,
) -> Result<ArrayRef> {
let opts = arrow_array::RecordBatchOptions::default().with_row_count(Some(vis.len()));
let input =
arrow_array::RecordBatch::try_new_with_options(self.arg_schema.clone(), columns, &opts)
.expect("failed to build record batch");
Expand All @@ -71,16 +99,8 @@ impl Expression for UdfExpression {
array.set_bitmap(array.null_bitmap() & vis);
Ok(Arc::new(array))
}

async fn eval_row(&self, input: &OwnedRow) -> Result<Datum> {
let chunk = DataChunk::from_rows(std::slice::from_ref(input), &self.arg_types);
let output_array = self.eval(&chunk).await?;
Ok(output_array.to_datum())
}
}

impl UdfExpression {}

#[cfg(not(madsim))]
impl<'a> TryFrom<&'a ExprNode> for UdfExpression {
type Error = ExprError;
Expand Down