diff --git a/cpp/src/arrow/acero/plan_test.cc b/cpp/src/arrow/acero/plan_test.cc index 2cb03114de6..ceabeeaf099 100644 --- a/cpp/src/arrow/acero/plan_test.cc +++ b/cpp/src/arrow/acero/plan_test.cc @@ -757,6 +757,30 @@ 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{})); + + 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); +} + 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 188a2708835..71abdc2a035 100644 --- a/cpp/src/arrow/acero/project_node.cc +++ b/cpp/src/arrow/acero/project_node.cc @@ -67,14 +67,24 @@ 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()); + bool nullable = true; + if (const auto* parameter = expr.parameter()) { + if (parameter->indices.size() == 1) { + int index = parameter->indices[0]; + if (index >= 0 && index < input_schema.num_fields()) { + nullable = input_schema.field(index)->nullable(); + } + } + } + fields[i] = field(std::move(names[i]), expr.type()->GetSharedPtr(), nullable); ++i; } return plan->EmplaceNode(plan, std::move(inputs),