Fix extract processor to not override errors with other status.

PiperOrigin-RevId: 715388381
Change-Id: Id66a6aba15252a4e6a12ed111b830b8226307b4c

Author: Koladata Team

koladata/internal/op_utils/BUILD

@@ -161,6 +161,7 @@ cc_test(
         "@com_google_absl//absl/log:check",
         "@com_google_absl//absl/status",
         "@com_google_absl//absl/status:status_matchers",
+        "@com_google_absl//absl/strings",
         "@com_google_absl//absl/types:span",
         "@com_google_arolla//arolla/dense_array",
         "@com_google_arolla//arolla/memory",

koladata/internal/op_utils/extract.cc

@@ -523,17 +523,16 @@ class CopyingProcessor {
       }
       return ProcessDicts(ds, db, fallbacks);
     }
-    absl::Status status = absl::OkStatus();
     for (const auto& attr_name_item : attr_names) {
       const std::string_view attr_name = attr_name_item.value<arolla::Text>();
       ASSIGN_OR_RETURN((auto [attr_schema, was_schema_updated]),
           CopyAttrSchema(ds.schema, db, fallbacks, attr_name));
       if (was_schema_updated || ds.slice.present_count() > 0) {
         // Data or schema are not yet copied.
-        status = ProcessAttribute(ds, attr_name, attr_schema);
+        RETURN_IF_ERROR(ProcessAttribute(ds, attr_name, attr_schema));
       }
     }
-    return status;
+    return absl::OkStatus();
   }
 
   // Process slice of objects with entity schema.

koladata/internal/op_utils/extract_test.cc

@@ -27,6 +27,7 @@
 #include "absl/log/check.h"
 #include "absl/status/status.h"
 #include "absl/status/status_matchers.h"
+#include "absl/strings/str_cat.h"
 #include "absl/types/span.h"
 #include "koladata/internal/data_bag.h"
 #include "koladata/internal/data_item.h"
@@ -2237,6 +2238,34 @@ TEST_P(ExtractTest, AnySchemaType) {
                "clone/extract not supported for kAny schema"));
 }
 
+TEST_P(ExtractTest, AnySchemaForOneAttribute) {
+  auto db = DataBagImpl::CreateEmptyDatabag();
+  auto obj_ids = AllocateEmptyObjects(3);
+  auto schema = AllocateSchema();
+
+  for (int i = 0; i < 15; ++i) {
+    SetSchemaTriples(
+        *db, {{schema, {{absl::StrCat("x", i), DataItem(schema::kInt32)}}}});
+  }
+  SetSchemaTriples(
+      *db, {{schema, {{absl::StrCat("x", 15), DataItem(schema::kAny)}}}});
+  for (int i = 0; i < 15; ++i) {
+    SetSchemaTriples(
+        *db, {{schema, {{absl::StrCat("x", i), DataItem(schema::kInt32)}}}});
+  }
+  for (int i = 0; i < 30; ++i) {
+    SetDataTriples(*db,
+                   {{obj_ids[0], {{absl::StrCat("x", i), obj_ids[1]}}}});
+  }
+
+  auto result_db = DataBagImpl::CreateEmptyDatabag();
+  EXPECT_THAT(
+      ExtractOp(result_db.get())(obj_ids, schema, *GetMainDb(db),
+                                 {GetFallbackDb(db).get()}, nullptr, {}),
+      StatusIs(absl::StatusCode::kInternal,
+               "clone/extract not supported for kAny schema"));
+}
+
 TEST_P(ExtractTest, AnySchemaTypeEmptySlice) {
   // TODO: expect error in test.
   auto db = DataBagImpl::CreateEmptyDatabag();