This is an automated email from the ASF dual-hosted git repository.
zanmato1984 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/main by this push:
new 5ef3c52003 GH-51231: [C++][Acero] Keep input nullability information
for direct field projections (#51691)
5ef3c52003 is described below
commit 5ef3c5200359bc8e2375daf5f989369b0c8c089b
Author: FadyYosry77 <[email protected]>
AuthorDate: Fri Oct 9 05:40:57 2026 +0300
GH-51231: [C++][Acero] Keep input nullability information for direct field
projections (#51691)
### Rationale for this change
ProjectNode creates output column definitions with the requested output
name and the type of the bound expression for that column. Nullability
information is available from the input schema but is not copied during output
column creation; thus, output columns are nullable by default.
As a result, projecting a nonnullable column from the input schema produces
a nullable column in the output schema, even when only that column is
projected. An identity projection may therefore produce a schema incompatible
with its input.
GH-51231 restricts this change to top-level input fields.
### What changes does this PR contain?
Nullability is now exposed through `Expression::nullable()`. Binding
captures the input field's nullability for direct top-level field references,
and ProjectNode consumes that accessor to determine the nullability of the
output field.
The captured nullability is recomputed when rebinding and preserved when
converting named references to field paths.
The projected field name and type of the bound expression remain unchanged.
Calls, literals, nested field references, and other expressions whose
nullability has not been inferred remain conservatively nullable. This pull
request does not include full function-aware nullability inference.
A regression test covers:
- direct nonnullable-nullable field correspondence
- reordered projections
- renaming a required field
- a calculated control with nonnullable input and nullable output as before
- already-bound indexed field projections
Expression-level regression tests cover direct fields, the conservative
fallback, rebinding, and nullability preservation when converting named
references to field paths.
### Are these changes tested?
Yes.
The regression test checks the schemas created. It failed with the old
implementation and passed with the new implementation.
Local validation included:
- targeted GH-51231 regression: 1/1 passed
- expression-level nullability regressions: 5/5 passed
- full arrow-compute-expression-test: 54/54 passed
- full arrow-acero-plan-test: 63/63 passed
- registered expression and plan CTests: 2/2 passed
- clang-format v18.1.8 passed
- cpplint 1.6.1 passed
- output of git diff --check and git diff --cached --check was clean
The targeted regression analysis was conducted with:
```bash
./cpp/build-gh51231/debug/arrow-acero-plan-test \
--gtest_filter=ExecPlanExecution.ProjectPreservesDirectFieldNullability
```
The expression and Acero plan test suites were evaluated with:
```bash
PYTHON=python3 ctest --test-dir cpp/build-gh51231 \
--output-on-failure \
-j2 \
-R '^(arrow-compute-expression-test|arrow-acero-plan-test)$'
```
Substrait integration was not built locally. The regression test is defined
at the native Acero ProjectNode level.
### Are there any visible user impacts?
Yes. Top-level field projections copy the nullable attribute from input
fields to output fields.
This change only alters the projected schema. Projected values remain the
same.
The experimental Expression API gains the conservative `nullable()`
accessor, and bound parameters store nullability.
### Was AI involved in this PR?
Yes. AI tools were used to assist with the investigation, implementation
and test drafting, review, and preparation of this description.
I reproduced the issue locally and reviewed the patch, verifying that the
test case failed without the patch and passed with the patch, and ran the
listed C++ tests and style checks.
**PR code and description by:**
- [x] Human
- [x] AI
**Reviewed before submission by:**
- [x] Human
- [x] AI
- [ ] Not reviewed
* GitHub Issue: #51231
Authored-by: Fady Youssery <[email protected]>
Signed-off-by: Rossi Sun <[email protected]>
---
cpp/src/arrow/acero/plan_test.cc | 32 ++++++++++++
cpp/src/arrow/acero/project_node.cc | 8 +--
cpp/src/arrow/compute/expression.cc | 18 +++++--
cpp/src/arrow/compute/expression.h | 9 +++-
cpp/src/arrow/compute/expression_test.cc | 84 ++++++++++++++++++++++++++++++++
5 files changed, 142 insertions(+), 9 deletions(-)
diff --git a/cpp/src/arrow/acero/plan_test.cc b/cpp/src/arrow/acero/plan_test.cc
index 2cb03114de..cee15f3427 100644
--- a/cpp/src/arrow/acero/plan_test.cc
+++ b/cpp/src/arrow/acero/plan_test.cc
@@ -757,6 +757,38 @@ TEST(ExecPlanExecution, DeclarationToSchema) {
AssertSchemaEqual(expected_out_schema, actual_out_schema);
}
+TEST(ExecPlanExecution, ProjectPreservesDirectFieldNullability) {
+ auto input_schema = schema({field("r", int64(), false), field("n", int64(),
true)});
+ Declaration source(
+ "exec_batch_source",
+ ExecBatchSourceNodeOptions(input_schema,
std::vector<compute::ExecBatch>{}));
+
+ auto direct = Declaration::Sequence(
+ {source,
+ {"project", ProjectNodeOptions({field_ref("r"), field_ref("n")}, {"r",
"n"})}});
+ ASSERT_OK_AND_ASSIGN(auto direct_schema, DeclarationToSchema(direct));
+ AssertSchemaEqual(input_schema, direct_schema);
+
+ auto reordered = Declaration::Sequence(
+ {source,
+ {"project",
+ ProjectNodeOptions({field_ref("n"), field_ref("r"),
+ call("add", {field_ref("r"),
literal(int64_t{1})})},
+ {"n", "renamed", "computed"})}});
+ ASSERT_OK_AND_ASSIGN(auto reordered_schema, DeclarationToSchema(reordered));
+ AssertSchemaEqual(schema({field("n", int64(), true), field("renamed",
int64(), false),
+ field("computed", int64(), true)}),
+ reordered_schema);
+
+ ASSERT_OK_AND_ASSIGN(auto bound_r, field_ref(0).Bind(*input_schema));
+ ASSERT_OK_AND_ASSIGN(auto bound_n, field_ref(1).Bind(*input_schema));
+ auto bound = Declaration::Sequence(
+ {source, {"project", ProjectNodeOptions({bound_n, bound_r}, {"n",
"renamed"})}});
+ ASSERT_OK_AND_ASSIGN(auto bound_schema, DeclarationToSchema(bound));
+ AssertSchemaEqual(schema({field("n", int64(), true), field("renamed",
int64(), false)}),
+ bound_schema);
+}
+
TEST(ExecPlanExecution, DeclarationToReader) {
auto basic_data = MakeBasicBatches();
auto plan = Declaration::Sequence(
diff --git a/cpp/src/arrow/acero/project_node.cc
b/cpp/src/arrow/acero/project_node.cc
index 188a270883..6ad7a59d73 100644
--- a/cpp/src/arrow/acero/project_node.cc
+++ b/cpp/src/arrow/acero/project_node.cc
@@ -67,14 +67,16 @@ class ProjectNode : public MapNode {
" doesn't match size of expressions " +
std::to_string(exprs.size())));
}
+ const auto& input_schema = *inputs[0]->output_schema();
FieldVector fields(exprs.size());
int i = 0;
for (auto& expr : exprs) {
if (!expr.IsBound()) {
- ARROW_ASSIGN_OR_RAISE(expr, expr.Bind(*inputs[0]->output_schema(),
-
plan->query_context()->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(
+ expr, expr.Bind(input_schema,
plan->query_context()->exec_context()));
}
- fields[i] = field(std::move(names[i]), expr.type()->GetSharedPtr());
+ fields[i] =
+ field(std::move(names[i]), expr.type()->GetSharedPtr(),
expr.nullable());
++i;
}
return plan->EmplaceNode<ProjectNode>(plan, std::move(inputs),
diff --git a/cpp/src/arrow/compute/expression.cc
b/cpp/src/arrow/compute/expression.cc
index 93427f75fe..cd95205566 100644
--- a/cpp/src/arrow/compute/expression.cc
+++ b/cpp/src/arrow/compute/expression.cc
@@ -123,6 +123,15 @@ const DataType* Expression::type() const {
return CallNotNull(*this)->type.type;
}
+bool Expression::nullable() const {
+ if (const Parameter* parameter = this->parameter()) {
+ if (parameter->type.type != nullptr) {
+ return parameter->nullable;
+ }
+ }
+ return true;
+}
+
namespace {
std::string PrintDatum(const Datum& datum) {
@@ -620,6 +629,7 @@ Result<Expression> BindImpl(Expression expr, const
TypeOrSchema& in,
std::copy(path.indices().begin(), path.indices().end(),
param.indices.begin());
ARROW_ASSIGN_OR_RAISE(auto field, path.Get(in));
param.type = field->type();
+ param.nullable = path.indices().size() == 1 ? field->nullable() : true;
return Expression{std::move(param)};
}
@@ -1487,10 +1497,10 @@ Result<Expression> RemoveNamedRefs(Expression src) {
[](Expression expr) {
const Expression::Parameter* param = expr.parameter();
if (param && !param->ref.IsFieldPath()) {
- FieldPath ref_as_path(
- std::vector<int>(param->indices.begin(), param->indices.end()));
- return Expression(
- Expression::Parameter{std::move(ref_as_path), param->type,
param->indices});
+ auto param_as_path = *param;
+ param_as_path.ref =
+ FieldPath(std::vector<int>(param->indices.begin(),
param->indices.end()));
+ return Expression(std::move(param_as_path));
}
return expr;
diff --git a/cpp/src/arrow/compute/expression.h
b/cpp/src/arrow/compute/expression.h
index b8ce50675c..c0dc6eadfc 100644
--- a/cpp/src/arrow/compute/expression.h
+++ b/cpp/src/arrow/compute/expression.h
@@ -115,8 +115,12 @@ class ARROW_EXPORT Expression {
/// The type to which this expression will evaluate
const DataType* type() const;
- // XXX someday
- // NullGeneralization::type nullable() const;
+
+ /// Whether this expression could evaluate to null.
+ /// Returns false only if the bound input guarantees a non-null result.
+ /// Currently, only bound top-level field references have inferred
nullability.
+ /// Returns true for other expressions, including unbound expressions.
+ bool nullable() const;
struct Parameter {
FieldRef ref;
@@ -124,6 +128,7 @@ class ARROW_EXPORT Expression {
// post-bind properties
TypeHolder type;
::arrow::internal::SmallVector<int, 2> indices;
+ bool nullable = true;
};
const Parameter* parameter() const;
diff --git a/cpp/src/arrow/compute/expression_test.cc
b/cpp/src/arrow/compute/expression_test.cc
index b4ae405b35..10c1b348e6 100644
--- a/cpp/src/arrow/compute/expression_test.cc
+++ b/cpp/src/arrow/compute/expression_test.cc
@@ -612,6 +612,67 @@ TEST(Expression, BindNestedFieldRef) {
field("b",
int64())}))})));
}
+TEST(Expression, NullableFieldRef) {
+ auto input_schema = schema({field("r", int32(), false), field("n", int32(),
true)});
+ for (int i = 0; i < input_schema->num_fields(); ++i) {
+ for (const auto& ref : {FieldRef(input_schema->field(i)->name()),
FieldRef(i)}) {
+ SCOPED_TRACE(input_schema->field(i)->ToString());
+ SCOPED_TRACE(ref.ToString());
+ auto expr = field_ref(ref);
+ EXPECT_TRUE(expr.nullable());
+
+ ASSERT_OK_AND_ASSIGN(auto bound, expr.Bind(*input_schema));
+ EXPECT_EQ(bound.nullable(), input_schema->field(i)->nullable());
+ EXPECT_TRUE(expr.nullable());
+
+ ASSERT_OK_AND_ASSIGN(auto bound_to_type,
+ expr.Bind(struct_(input_schema->fields())));
+ EXPECT_EQ(bound_to_type.nullable(), bound.nullable());
+ }
+ }
+}
+
+TEST(Expression, NullableConservativeFallback) {
+ auto input_schema = schema({field("r", int32(), false), field("n", int32(),
true)});
+ EXPECT_TRUE(Expression{}.nullable());
+ for (const auto& expr : {literal(1),
literal(std::make_shared<Int32Scalar>()),
+ add(field_ref("r"), literal(1)),
is_valid(field_ref("n"))}) {
+ SCOPED_TRACE(expr.ToString());
+ EXPECT_TRUE(expr.nullable());
+ ASSERT_OK_AND_ASSIGN(auto bound, expr.Bind(*input_schema));
+ EXPECT_TRUE(bound.IsBound());
+ EXPECT_TRUE(bound.nullable());
+ }
+}
+
+TEST(Expression, NullableNestedFieldRef) {
+ for (bool parent_nullable : {false, true}) {
+ for (bool child_nullable : {false, true}) {
+ auto input_schema = schema(
+ {field("a", struct_({field("b", int32(), child_nullable)}),
parent_nullable)});
+ for (const auto& ref : {FieldRef("a", "b"), FieldRef(FieldPath({0,
0}))}) {
+ SCOPED_TRACE(input_schema->ToString());
+ SCOPED_TRACE(ref.ToString());
+ ASSERT_OK_AND_ASSIGN(auto bound, field_ref(ref).Bind(*input_schema));
+ EXPECT_TRUE(bound.nullable());
+ }
+ }
+ }
+}
+
+TEST(Expression, NullableRebind) {
+ auto required_schema = schema({field("a", int32(), false)});
+ auto optional_schema = schema({field("a", int32(), true)});
+ ASSERT_OK_AND_ASSIGN(auto required, field_ref("a").Bind(*required_schema));
+ ASSERT_OK_AND_ASSIGN(auto optional, required.Bind(*optional_schema));
+ EXPECT_FALSE(required.nullable());
+ EXPECT_TRUE(optional.nullable());
+
+ ASSERT_OK_AND_ASSIGN(auto rebound_required, optional.Bind(*required_schema));
+ EXPECT_FALSE(rebound_required.nullable());
+ EXPECT_TRUE(optional.nullable());
+}
+
TEST(Expression, BindCall) {
auto expr = add(field_ref("i32"), field_ref("i32_req"));
EXPECT_FALSE(expr.IsBound());
@@ -1401,6 +1462,29 @@ TEST(Expression, RemoveNamedRefs) {
ExpectRemovesRefsTo(field_ref({"a", "b"}), field_ref({0, 0}), nested_schema);
}
+TEST(Expression, NullableRemoveNamedRefs) {
+ ASSERT_OK_AND_ASSIGN(auto bound, field_ref("i32_req").Bind(*kBoringSchema));
+ ASSERT_OK_AND_ASSIGN(auto without_named_refs, RemoveNamedRefs(bound));
+ EXPECT_TRUE(without_named_refs.IsBound());
+ EXPECT_TRUE(without_named_refs.field_ref()->IsFieldPath());
+ EXPECT_FALSE(without_named_refs.nullable());
+ EXPECT_FALSE(bound.nullable());
+
+ ASSERT_OK_AND_ASSIGN(auto bound_call,
+ add(field_ref("i32_req"),
literal(1)).Bind(*kBoringSchema));
+ ASSERT_OK_AND_ASSIGN(auto without_named_refs_call,
RemoveNamedRefs(bound_call));
+ EXPECT_TRUE(without_named_refs_call.IsBound());
+ EXPECT_TRUE(without_named_refs_call.nullable());
+ const auto* call = without_named_refs_call.call();
+ ASSERT_NE(call, nullptr);
+ ASSERT_EQ(call->arguments.size(), 2);
+ const auto& field_arg = call->arguments[0];
+ EXPECT_TRUE(field_arg.IsBound());
+ ASSERT_NE(field_arg.field_ref(), nullptr);
+ EXPECT_TRUE(field_arg.field_ref()->IsFieldPath());
+ EXPECT_FALSE(field_arg.nullable());
+}
+
TEST(Expression, ExtractKnownFieldValues) {
struct {
void operator()(Expression guarantee,
