fix arrow memory leak

Author: xiao.dong

src/iceberg/CMakeLists.txt

@@ -45,6 +45,7 @@ set(ICEBERG_SOURCES
     type.cc
     manifest_reader.cc
     manifest_reader_internal.cc
+    arrow_struct_guard.cc
     util/murmurhash3_internal.cc
     util/timepoint.cc
     util/unreachable.cc

src/iceberg/arrow_struct_guard.cc

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/arrow_struct_guard.h"
+
+namespace iceberg::internal {
+
+ArrowArrayGuard::~ArrowArrayGuard() {
+  if (array_ != nullptr) {
+    ArrowArrayRelease(array_);
+  }
+}
+
+ArrowSchemaGuard::~ArrowSchemaGuard() {
+  if (schema_ != nullptr) {
+    ArrowSchemaRelease(schema_);
+  }
+}
+
+ArrowArrayViewGuard::~ArrowArrayViewGuard() {
+  if (view_ != nullptr) {
+    ArrowArrayViewReset(view_);
+  }
+}
+
+ArrowArrayBufferGuard::~ArrowArrayBufferGuard() {
+  if (buffer_ != nullptr) {
+    ArrowBufferReset(buffer_);
+  }
+}
+
+}  // namespace iceberg::internal

src/iceberg/arrow_struct_guard.h

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data.h"
+
+namespace iceberg::internal {
+
+class ArrowArrayGuard {
+ public:
+  ArrowArrayGuard(ArrowArray* array) : array_(array) {}
+  ~ArrowArrayGuard();
+
+ private:
+  ArrowArray* array_;
+};
+
+class ArrowSchemaGuard {
+ public:
+  ArrowSchemaGuard(ArrowSchema* schema) : schema_(schema) {}
+  ~ArrowSchemaGuard();
+
+ private:
+  ArrowSchema* schema_;
+};
+
+class ArrowArrayViewGuard {
+ public:
+  ArrowArrayViewGuard(ArrowArrayView* view) : view_(view) {}
+  ~ArrowArrayViewGuard();
+
+ private:
+  ArrowArrayView* view_;
+};
+
+class ArrowArrayBufferGuard {
+ public:
+  ArrowArrayBufferGuard(ArrowBuffer* buffer) : buffer_(buffer) {}
+  ~ArrowArrayBufferGuard();
+
+ private:
+  ArrowBuffer* buffer_;
+};
+
+}  // namespace iceberg::internal

src/iceberg/manifest_reader_internal.cc

@@ -23,6 +23,7 @@
 
 #include <nanoarrow/nanoarrow.h>
 
+#include "iceberg/arrow_struct_guard.h"
 #include "iceberg/manifest_entry.h"
 #include "iceberg/manifest_list.h"
 #include "iceberg/schema.h"
@@ -51,6 +52,7 @@ Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
   ArrowArrayView array_view;
   auto status = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
   ARROW_RETURN_IF_NOT_OK(status, error);
+  internal::ArrowArrayViewGuard view_guard(&array_view);
   status = ArrowArrayViewSetArray(&array_view, array_in, &error);
   ARROW_RETURN_IF_NOT_OK(status, error);
   status = ArrowArrayViewValidate(&array_view, NANOARROW_VALIDATION_LEVEL_FULL, &error);
@@ -65,8 +67,6 @@ Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
   for (int64_t idx = 0; idx < array_in->n_children; idx++) {
     const auto& field = iceberg_schema.GetFieldByIndex(idx);
     if (!field.has_value()) {
-      ArrowArrayRelease(array_in);
-      ArrowArrayViewReset(&array_view);
       return InvalidArgument("Field not found in schema: {}", idx);
     }
     auto field_name = field.value().get().name();
@@ -197,8 +197,6 @@ Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
     }
   }
 #undef PARSE_PRIMITIVE_FIELD
-  ArrowArrayRelease(array_in);
-  ArrowArrayViewReset(&array_view);
   return manifest_files;
 }  // namespace iceberg
 
@@ -213,6 +211,7 @@ Result<std::vector<std::unique_ptr<ManifestFile>>> ManifestListReaderImpl::Files
     return InvalidArgument("Get schema failed in reader:{}",
                            arrow_schema.error().message);
   }
+  internal::ArrowSchemaGuard schema_guard(&arrow_schema.value());
   auto schema = FromArrowSchema(arrow_schema.value(), std::nullopt);
   if (!schema.has_value()) {
     return InvalidArgument("Parse iceberg schema failed:{}", schema.error().message);
@@ -224,6 +223,7 @@ Result<std::vector<std::unique_ptr<ManifestFile>>> ManifestListReaderImpl::Files
                              result.error().message);
     }
     if (result.value().has_value()) {
+      internal::ArrowArrayGuard array_guard(&result.value().value());
       auto parse_result = ParseManifestListEntry(
           &arrow_schema.value(), &result.value().value(), *schema.value());
       if (!parse_result.has_value()) {
@@ -234,6 +234,7 @@ Result<std::vector<std::unique_ptr<ManifestFile>>> ManifestListReaderImpl::Files
                             std::make_move_iterator(parse_result.value().begin()),
                             std::make_move_iterator(parse_result.value().end()));
     } else {
+      // eof
       break;
     }
   }