GH-44318: [C++][Python] Fix RecordBatch::FromStructArray for sliced arrays with offset = 0

Author: wlhjason

cpp/src/arrow/array/array_nested.cc

@@ -1134,7 +1134,8 @@ Result<std::shared_ptr<Array>> StructArray::GetFlattenedField(int index,
   std::shared_ptr<Buffer> flattened_null_bitmap;
   int64_t flattened_null_count = kUnknownNullCount;
 
-  // Need to adjust for parent offset
+  // Push any non-trivial slicing on the parent to the child
+  // (including cases with offset = 0)
   if (data_->offset != 0 || data_->length != child_data->length) {
     child_data = child_data->Slice(data_->offset, data_->length);
   }

cpp/src/arrow/record_batch.cc

@@ -257,18 +257,13 @@ Result<std::shared_ptr<RecordBatch>> RecordBatch::FromStructArray(
     return Status::TypeError("Cannot construct record batch from array of type ",
                              *array->type());
   }
-  if (array->null_count() != 0 || array->offset() != 0) {
-    // If the struct array has a validity map or offset we need to push those into
-    // the child arrays via Flatten since the RecordBatch doesn't have validity/offset
-    const std::shared_ptr<StructArray>& struct_array =
-        internal::checked_pointer_cast<StructArray>(array);
-    ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Array>> fields,
-                          struct_array->Flatten(memory_pool));
-    return Make(arrow::schema(array->type()->fields()), array->length(),
-                std::move(fields));
-  }
-  return Make(arrow::schema(array->type()->fields()), array->length(),
-              array->data()->child_data);
+  // Push the struct array's validity map and slicing (if any) into the child arrays
+  // by calling Flatten
+  const std::shared_ptr<StructArray>& struct_array =
+      internal::checked_pointer_cast<StructArray>(array);
+  ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Array>> fields,
+                        struct_array->Flatten(memory_pool));
+  return Make(arrow::schema(array->type()->fields()), array->length(), std::move(fields));
 }
 
 namespace {

cpp/src/arrow/record_batch_test.cc

@@ -614,13 +614,28 @@ TEST_F(TestRecordBatch, FromSlicedStructArray) {
   StructArray struct_array(struct_({field("x", int64())}), kLength, {x_arr});
   std::shared_ptr<Array> sliced = struct_array.Slice(5, 3);
   ASSERT_OK_AND_ASSIGN(auto batch, RecordBatch::FromStructArray(sliced));
+  ASSERT_OK(batch->ValidateFull());
 
   std::shared_ptr<Array> expected_arr = ArrayFromJSON(int64(), "[5, 6, 7]");
   std::shared_ptr<RecordBatch> expected =
       RecordBatch::Make(schema({field("x", int64())}), 3, {expected_arr});
   AssertBatchesEqual(*expected, *batch);
 }
 
+TEST_F(TestRecordBatch, FromSlicedStructArrayWithOffsetZero) {
+  static constexpr int64_t kLength = 5;
+  std::shared_ptr<Array> x_arr = ArrayFromJSON(int64(), "[0, 1, 2, 3, 4]");
+  StructArray struct_array(struct_({field("x", int64())}), kLength, {x_arr});
+  std::shared_ptr<Array> sliced = struct_array.Slice(0, 3);
+  ASSERT_OK_AND_ASSIGN(auto batch, RecordBatch::FromStructArray(sliced));
+  ASSERT_OK(batch->ValidateFull());
+
+  std::shared_ptr<Array> expected_arr = ArrayFromJSON(int64(), "[0, 1, 2]");
+  std::shared_ptr<RecordBatch> expected =
+      RecordBatch::Make(schema({field("x", int64())}), 3, {expected_arr});
+  AssertBatchesEqual(*expected, *batch);
+}
+
 TEST_F(TestRecordBatch, FromStructArrayInvalidType) {
   random::RandomArrayGenerator gen(42);
   ASSERT_RAISES(TypeError, RecordBatch::FromStructArray(gen.ArrayOf(int32(), 6)));