You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "dongxiao1198 (via GitHub)" <gi...@apache.org> on 2025/07/08 03:24:59 UTC
[PR] feat: add manifest list reader [iceberg-cpp]
dongxiao1198 opened a new pull request, #143:
URL: https://github.com/apache/iceberg-cpp/pull/143
- Add manifest list reader
- Integrate with avro reader
- Add simple ut
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2192006916
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) { \
+ return InvalidArrowData("NanoArrow error: {}", error.message); \
Review Comment:
```suggestion
return InvalidArrowData("Nanoarrow error: {}", error.message); \
```
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) { \
Review Comment:
```suggestion
if (status != NANOARROW_OK) [[unlikely]] { \
```
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) { \
+ return InvalidArrowData("NanoArrow error: {}", error.message); \
+ }
+
+Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
+ ArrowSchema* schema, ArrowArray* array_in, const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<std::unique_ptr<ManifestFile>> manifest_files;
+ manifest_files.resize(array_in->length);
+ for (auto& manifest_file : manifest_files) {
+ manifest_file = std::make_unique<ManifestFile>();
+ }
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx]->field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx]->manifest_path = path_str;
+ }
+ }
+ } else if (field_name == ManifestFile::kManifestLength.name()) {
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ } else if (field_name == ManifestFile::kPartitionSpecId.name()) {
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ } else if (field_name == ManifestFile::kContent.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx]->content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ } else if (field_name == ManifestFile::kSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kMinSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kAddedSnapshotId.name()) {
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ } else if (field_name == ManifestFile::kAddedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ } else if (field_name == ManifestFile::kExistingFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ } else if (field_name == ManifestFile::kDeletedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ } else if (field_name == ManifestFile::kAddedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kExistingRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kDeletedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kPartitions.name()) {
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ auto manifest_count = view_of_column->length;
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidArgument("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidArgument("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidArgument("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidArgument("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidArgument("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidArgument("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidArgument("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset =
+ ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset;
+ partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file->partitions.emplace_back(partition_field_summary);
+ }
+ }
+ } else if (field_name == ManifestFile::kKeyMetadata.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx]->key_metadata = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ }
+ } else if (field_name == ManifestFile::kFirstRowId.name()) {
+ PARSE_PRIMITIVE_FIELD(first_row_id, int64_t);
+ } else {
+ return InvalidArgument("Unsupported type: {}", field_name);
+ }
+ }
+#undef PARSE_PRIMITIVE_FIELD
+ return manifest_files;
+} // namespace iceberg
+
+Result<std::vector<std::unique_ptr<ManifestEntry>>> ManifestReaderImpl::Entries() const {
+ return {};
+}
+
+Result<std::vector<std::unique_ptr<ManifestFile>>> ManifestListReaderImpl::Files() const {
+ std::vector<std::unique_ptr<ManifestFile>> manifest_files;
+ auto arrow_schema = reader_->Schema();
+ if (!arrow_schema.has_value()) {
+ 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);
Review Comment:
We should pass this schema explicitly when creating the reader instead of converting it back here.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) { \
+ return InvalidArrowData("NanoArrow error: {}", error.message); \
+ }
+
+Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
+ ArrowSchema* schema, ArrowArray* array_in, const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<std::unique_ptr<ManifestFile>> manifest_files;
+ manifest_files.resize(array_in->length);
+ for (auto& manifest_file : manifest_files) {
+ manifest_file = std::make_unique<ManifestFile>();
+ }
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx]->field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
Review Comment:
The schemas to read from manifest list and manifest file are provided by the library so they are fixed. We can simply define a `map<string, size_t>` to find the index of a specific field_name.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) { \
+ return InvalidArrowData("NanoArrow error: {}", error.message); \
+ }
+
+Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
+ ArrowSchema* schema, ArrowArray* array_in, const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<std::unique_ptr<ManifestFile>> manifest_files;
+ manifest_files.resize(array_in->length);
+ for (auto& manifest_file : manifest_files) {
+ manifest_file = std::make_unique<ManifestFile>();
+ }
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx]->field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx]->manifest_path = path_str;
+ }
+ }
+ } else if (field_name == ManifestFile::kManifestLength.name()) {
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ } else if (field_name == ManifestFile::kPartitionSpecId.name()) {
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ } else if (field_name == ManifestFile::kContent.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx]->content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ } else if (field_name == ManifestFile::kSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kMinSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kAddedSnapshotId.name()) {
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ } else if (field_name == ManifestFile::kAddedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ } else if (field_name == ManifestFile::kExistingFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ } else if (field_name == ManifestFile::kDeletedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ } else if (field_name == ManifestFile::kAddedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kExistingRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kDeletedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kPartitions.name()) {
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ auto manifest_count = view_of_column->length;
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidArgument("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidArgument("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidArgument("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidArgument("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidArgument("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidArgument("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidArgument("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset =
+ ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset;
+ partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file->partitions.emplace_back(partition_field_summary);
+ }
+ }
+ } else if (field_name == ManifestFile::kKeyMetadata.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx]->key_metadata = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ }
+ } else if (field_name == ManifestFile::kFirstRowId.name()) {
+ PARSE_PRIMITIVE_FIELD(first_row_id, int64_t);
+ } else {
+ return InvalidArgument("Unsupported type: {}", field_name);
+ }
+ }
+#undef PARSE_PRIMITIVE_FIELD
+ return manifest_files;
+} // namespace iceberg
+
+Result<std::vector<std::unique_ptr<ManifestEntry>>> ManifestReaderImpl::Entries() const {
+ return {};
+}
+
+Result<std::vector<std::unique_ptr<ManifestFile>>> ManifestListReaderImpl::Files() const {
+ std::vector<std::unique_ptr<ManifestFile>> manifest_files;
+ auto arrow_schema = reader_->Schema();
+ if (!arrow_schema.has_value()) {
+ return InvalidArgument("Get schema failed in reader:{}",
+ arrow_schema.error().message);
+ }
Review Comment:
```suggestion
ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, reader_->Schema());
```
nit: you may simply write the above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2203968566
##########
test/manifest_list_reader_test.cc:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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 <arrow/filesystem/localfs.h>
+#include <avro/GenericDatum.hh>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_reader.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_reader.h"
+#include "iceberg/schema.h"
+#include "matchers.h"
+#include "temp_file_test_base.h"
+#include "test_common.h"
+
+namespace iceberg {
+
+class ManifestListReaderTest : public TempFileTestBase {
+ protected:
+ static void SetUpTestSuite() { avro::AvroReader::Register(); }
+
+ void SetUp() override {
+ TempFileTestBase::SetUp();
+ local_fs_ = std::make_shared<::arrow::fs::LocalFileSystem>();
+ file_io_ = std::make_shared<iceberg::arrow::ArrowFileSystemFileIO>(local_fs_);
+ }
+
+ std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;
+ std::shared_ptr<FileIO> file_io_;
+};
+
+TEST_F(ManifestListReaderTest, BasicTest) {
Review Comment:
got it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2191968847
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
BTW, should we create the reader internally using ReaderFactory? We know the manifest schema and file format. Sometimes it is useful to directly return Arrow arrays of manifest entries. Therefore I propose the following interface and make them all static functions:
```cpp
class ICEBERG_EXPORT ManifestReader {
public:
static Result<std::unique_ptr<Reader>> MakeReader(
const std::string& manifest_list_location,
FileIO& file_io,
std::shared_ptr<Schema> schema = kManifestSchema);
static Result<std::vector<std::unique_ptr<ManifestEntry>>> ReadEntries(
const std::string& manifest_list_location,
FileIO& file_io);
};
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#issuecomment-3067706152
Could you help review this? @lidavidm @zhjwpku
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2206219457
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define NANOARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Status ParsePartitionFieldSummaryList(ArrowArrayView* view_of_column,
+ std::vector<ManifestFile>& manifest_files) {
+ auto manifest_count = view_of_column->length;
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidManifestList("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidManifestList("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidManifestList("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset; partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file.partitions.emplace_back(partition_field_summary);
+ }
+ }
+ return {};
+}
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ ArrowArrayView array_view;
+ auto status = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ internal::ArrowArrayViewGuard view_guard(&array_view);
+ status = ArrowArrayViewSetArray(&array_view, array_in, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ status = ArrowArrayViewValidate(&array_view, NANOARROW_VALIDATION_LEVEL_FULL, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidSchema("Field index {} is not found in schema", idx);
+ }
+ auto field_name = field.value().get().name();
+ bool required = !field.value().get().optional();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(item, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
Review Comment:
maybe specific the version of manifest&manifest list in v1|v2|v3 is better to valid the schema of file
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo merged PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2203935650
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
Review Comment:
we got only 3 getters for primitive types from nanoarrow
ArrowArrayViewGetIntUnsafe
ArrowArrayViewGetUIntUnsafe
ArrowArrayViewGetDoubleUnsafe
since manifest has no float and uint, it is ok here
nanoarrow cast all int type(int32 int64 int8) to int64..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2206155534
##########
src/iceberg/manifest_reader.h:
##########
@@ -35,35 +35,32 @@ namespace iceberg {
class ICEBERG_EXPORT ManifestReader {
public:
virtual ~ManifestReader() = default;
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::unique_ptr<ManifestReader>> MakeReader(
+ const std::string& manifest_location, std::shared_ptr<FileIO> file_io,
+ std::shared_ptr<Schema> partition_schema);
};
/// \brief Read manifest files from a manifest list file.
class ICEBERG_EXPORT ManifestListReader {
public:
virtual ~ManifestListReader() = default;
- virtual Result<std::span<std::unique_ptr<ManifestFile>>> Files() const = 0;
+ virtual Result<std::vector<ManifestFile>> Files() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::unique_ptr<ManifestListReader>> MakeReader(
+ const std::string& manifest_list_location, std::shared_ptr<FileIO> file_io);
};
/// \brief Creates a reader for the manifest list.
/// \param file_path Path to the manifest list file.
/// \return A Result containing the reader or an error.
Result<std::unique_ptr<ManifestListReader>> CreateManifestListReader(
- std::string_view file_path) {
- return NotImplemented("CreateManifestListReader is not implemented yet.");
-}
+ std::string_view file_path);
Review Comment:
We need to remove `CreateManifestListReader` and `CreateManifestReader`, then use above static functions in the table_scan.cc as they are used there.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define NANOARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Status ParsePartitionFieldSummaryList(ArrowArrayView* view_of_column,
+ std::vector<ManifestFile>& manifest_files) {
+ auto manifest_count = view_of_column->length;
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidManifestList("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidManifestList("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidManifestList("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset; partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file.partitions.emplace_back(partition_field_summary);
+ }
+ }
+ return {};
+}
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ ArrowArrayView array_view;
+ auto status = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ internal::ArrowArrayViewGuard view_guard(&array_view);
+ status = ArrowArrayViewSetArray(&array_view, array_in, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ status = ArrowArrayViewValidate(&array_view, NANOARROW_VALIDATION_LEVEL_FULL, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidSchema("Field index {} is not found in schema", idx);
+ }
+ auto field_name = field.value().get().name();
+ bool required = !field.value().get().optional();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(item, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].item = static_cast<type>(value); \
+ } else if (required) { \
+ return InvalidManifestList("Field {} is required but null at row {}", field_name, \
+ row_idx); \
+ } \
+ }
+
+ switch (idx) {
+ case 0:
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx].manifest_path = path_str;
+ }
+ }
+ break;
+ case 1:
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ break;
+ case 2:
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ break;
+ case 3:
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ break;
+ case 4:
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ break;
+ case 5:
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ break;
+ case 6:
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ break;
+ case 7:
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ break;
+ case 8:
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ break;
+ case 9:
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ break;
+ case 10:
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ break;
+ case 11:
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ break;
+ case 12:
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ break;
+ case 13:
+ ICEBERG_RETURN_UNEXPECTED(
+ ParsePartitionFieldSummaryList(view_of_column, manifest_files));
+ break;
+ case 14:
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].key_metadata = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ }
+ break;
+ case 15:
+ PARSE_PRIMITIVE_FIELD(first_row_id, int64_t);
+ break;
+ default:
+ return InvalidManifestList("Unsupported type: {}", field_name);
+ }
+ }
+ return manifest_files;
+} // namespace iceberg
Review Comment:
```suggestion
}
```
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define NANOARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Status ParsePartitionFieldSummaryList(ArrowArrayView* view_of_column,
+ std::vector<ManifestFile>& manifest_files) {
+ auto manifest_count = view_of_column->length;
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidManifestList("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidManifestList("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidManifestList("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset; partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file.partitions.emplace_back(partition_field_summary);
+ }
+ }
+ return {};
+}
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ ArrowArrayView array_view;
+ auto status = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ internal::ArrowArrayViewGuard view_guard(&array_view);
+ status = ArrowArrayViewSetArray(&array_view, array_in, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ status = ArrowArrayViewValidate(&array_view, NANOARROW_VALIDATION_LEVEL_FULL, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidSchema("Field index {} is not found in schema", idx);
+ }
+ auto field_name = field.value().get().name();
+ bool required = !field.value().get().optional();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(item, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
Review Comment:
There is an issue with regard to the definition of `ManifestFile`. If a field is optional, we need to choose from the below options:
1. Define it as `std::optional<T>` in the `ManifestFile`, then we don't need to do anything if the read value is null. For example, `added_files_count` is defined in this way.
2. Define it as `T`, then we need to assign a default value depending on its meaning when the read value is null. For example, `content` and `sequence_number` are defined in this way.
I think we need to address this to avoid any potential headache in the future. We can fix this in a followup PR.
Any suggestion? @lidavidm @zhjwpku @gty404 @mapleFU
##########
src/iceberg/manifest_reader.h:
##########
@@ -35,35 +35,32 @@ namespace iceberg {
class ICEBERG_EXPORT ManifestReader {
public:
virtual ~ManifestReader() = default;
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::unique_ptr<ManifestReader>> MakeReader(
+ const std::string& manifest_location, std::shared_ptr<FileIO> file_io,
+ std::shared_ptr<Schema> partition_schema);
};
/// \brief Read manifest files from a manifest list file.
class ICEBERG_EXPORT ManifestListReader {
public:
virtual ~ManifestListReader() = default;
- virtual Result<std::span<std::unique_ptr<ManifestFile>>> Files() const = 0;
+ virtual Result<std::vector<ManifestFile>> Files() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::unique_ptr<ManifestListReader>> MakeReader(
+ const std::string& manifest_list_location, std::shared_ptr<FileIO> file_io);
Review Comment:
```suggestion
std::string_view manifest_list_location, std::shared_ptr<FileIO> file_io);
```
##########
src/iceberg/manifest_reader.h:
##########
@@ -35,35 +35,32 @@ namespace iceberg {
class ICEBERG_EXPORT ManifestReader {
public:
virtual ~ManifestReader() = default;
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::unique_ptr<ManifestReader>> MakeReader(
+ const std::string& manifest_location, std::shared_ptr<FileIO> file_io,
Review Comment:
```suggestion
std::string_view manifest_location, std::shared_ptr<FileIO> file_io,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2192050716
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
Agree, user should not care of creating the file reader or stream. I will make a refactor of those code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2192012781
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) { \
+ return InvalidArrowData("NanoArrow error: {}", error.message); \
+ }
+
+Result<std::vector<std::unique_ptr<ManifestFile>>> ParseManifestListEntry(
+ ArrowSchema* schema, ArrowArray* array_in, const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<std::unique_ptr<ManifestFile>> manifest_files;
+ manifest_files.resize(array_in->length);
+ for (auto& manifest_file : manifest_files) {
+ manifest_file = std::make_unique<ManifestFile>();
+ }
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx]->field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
Review Comment:
The schemas to read from manifest list and manifest file are provided by the library so they are fixed. We can simply define a `map<string, size_t>` to find the index of a specific field_name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2192189472
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
> For ReadEntries, I also think `std::vector<ManifestEntry>` looks better than `std::vector<std::unique_ptr<ManifestEntry>>`.
Manifest list schema is almost a fixed schema, i think there is no need to add it to interface
static Result<std::shared_ptr<ManifestListReader>> MakeReader(
const std::string& manifest_list_location, std::shared_ptr<FileIO> file_io);
But for manifest EnteryReader , the schema was based on partition type, i think the way like this is much better:
static Result<std::shared_ptr<ManifestReader>> MakeReader(
const std::string& manifest_location, std::shared_ptr<FileIO> file_io,
std::shared_ptr<Schema> partition_schema);
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2203726170
##########
src/iceberg/manifest_reader_internal.h:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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
+
+/// \file iceberg/internal/manifest_reader_internal.h
+/// Reader implement for manifest files.
Review Comment:
```suggestion
/// Reader implementation for manifest list files and manifest files.
```
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
Review Comment:
```suggestion
return InvalidSchema("Field index {} is not found in schema", idx);
```
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx].manifest_path = path_str;
+ }
+ }
+ } else if (field_name == ManifestFile::kManifestLength.name()) {
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ } else if (field_name == ManifestFile::kPartitionSpecId.name()) {
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ } else if (field_name == ManifestFile::kContent.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ } else if (field_name == ManifestFile::kSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kMinSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kAddedSnapshotId.name()) {
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ } else if (field_name == ManifestFile::kAddedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ } else if (field_name == ManifestFile::kExistingFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ } else if (field_name == ManifestFile::kDeletedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ } else if (field_name == ManifestFile::kAddedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kExistingRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kDeletedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kPartitions.name()) {
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ auto manifest_count = view_of_column->length;
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidArgument("partitions field should be a list.");
Review Comment:
`InvalidArgument` usually deals with invalid user input to a function. We might need to use `InvalidSchema` or even add a new `InvalidManifest` to be more specific. Same for other places.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx].manifest_path = path_str;
+ }
+ }
+ } else if (field_name == ManifestFile::kManifestLength.name()) {
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ } else if (field_name == ManifestFile::kPartitionSpecId.name()) {
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ } else if (field_name == ManifestFile::kContent.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ } else if (field_name == ManifestFile::kSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kMinSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kAddedSnapshotId.name()) {
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ } else if (field_name == ManifestFile::kAddedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ } else if (field_name == ManifestFile::kExistingFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ } else if (field_name == ManifestFile::kDeletedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ } else if (field_name == ManifestFile::kAddedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kExistingRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kDeletedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kPartitions.name()) {
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ auto manifest_count = view_of_column->length;
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidArgument("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidArgument("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidArgument("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidArgument("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidArgument("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidArgument("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type !=
+ ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidArgument("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset =
+ ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset;
+ partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file.partitions.emplace_back(partition_field_summary);
+ }
+ }
+ } else if (field_name == ManifestFile::kKeyMetadata.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].key_metadata = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ }
+ } else if (field_name == ManifestFile::kFirstRowId.name()) {
+ PARSE_PRIMITIVE_FIELD(first_row_id, int64_t);
+ } else {
+ return InvalidArgument("Unsupported type: {}", field_name);
+ }
+ }
+#undef PARSE_PRIMITIVE_FIELD
+ return manifest_files;
+} // namespace iceberg
+
+Result<std::vector<ManifestEntry>> ManifestReaderImpl::Entries() const { return {}; }
+
+Result<std::vector<ManifestFile>> ManifestListReaderImpl::Files() const {
+ std::vector<ManifestFile> manifest_files;
+ ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, reader_->Schema());
+ internal::ArrowSchemaGuard schema_guard(&arrow_schema);
+ while (true) {
+ auto result = reader_->Next();
+ if (!result.has_value()) {
+ return InvalidArgument("Failed to read manifest list entry:{}",
+ result.error().message);
+ }
+ if (result.value().has_value()) {
+ internal::ArrowArrayGuard array_guard(&result.value().value());
+ auto parse_result =
+ ParseManifestListEntry(&arrow_schema, &result.value().value(), *schema_);
+ if (!parse_result.has_value()) {
+ return InvalidArgument("Failed to parse manifest list entry:{}",
+ parse_result.error().message);
+ }
+ manifest_files.insert(manifest_files.end(),
+ std::make_move_iterator(parse_result.value().begin()),
+ std::make_move_iterator(parse_result.value().end()));
Review Comment:
```suggestion
ICEBERG_ASSIGN_OR_RAISE(auto next_data, reader_->Next());
if (next_data.has_value()) {
auto arrow_data = std::move(next_data.value());
internal::ArrowArrayGuard array_guard(&arrow_data);
ICEBERG_ASSIGN_OR_RAISE(auto entries,
ParseManifestListEntry(&arrow_schema, &arrow_data, *schema_));
manifest_files.insert(manifest_files.end(),
std::make_move_iterator(entries.begin()),
std::make_move_iterator(entries.end()));
```
I don't these errors should be wrapped as `InvalidArgument`. Let's use `ICEBERG_ASSIGN_OR_RAISE` to simplify them.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
Review Comment:
Please also include the error code in the message.
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,20 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::shared_ptr<ManifestReader>> MakeReader(
+ const std::string& manifest_location, std::shared_ptr<FileIO> file_io,
+ std::shared_ptr<Schema> partition_schema);
};
/// \brief Read manifest files from a manifest list file.
class ICEBERG_EXPORT ManifestListReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestFile>>> Files() const = 0;
+ virtual Result<std::vector<ManifestFile>> Files() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::shared_ptr<ManifestListReader>> MakeReader(
Review Comment:
```suggestion
static Result<std::unique_ptr<ManifestListReader>> MakeReader(
```
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
Review Comment:
Can we use template (together with concepts) instead of marco to be type safe? This marco only applies to integer values due to `ArrowArrayViewGetIntUnsafe`.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
Review Comment:
```suggestion
#define NANOARROW_RETURN_IF_NOT_OK(status, error) \
```
##########
test/manifest_list_reader_test.cc:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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 <arrow/filesystem/localfs.h>
+#include <avro/GenericDatum.hh>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_reader.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_reader.h"
+#include "iceberg/schema.h"
+#include "matchers.h"
+#include "temp_file_test_base.h"
+#include "test_common.h"
+
+namespace iceberg {
+
+class ManifestListReaderTest : public TempFileTestBase {
+ protected:
+ static void SetUpTestSuite() { avro::AvroReader::Register(); }
+
+ void SetUp() override {
+ TempFileTestBase::SetUp();
+ local_fs_ = std::make_shared<::arrow::fs::LocalFileSystem>();
+ file_io_ = std::make_shared<iceberg::arrow::ArrowFileSystemFileIO>(local_fs_);
+ }
+
+ std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;
+ std::shared_ptr<FileIO> file_io_;
+};
+
+TEST_F(ManifestListReaderTest, BasicTest) {
Review Comment:
We need to add `operator==` overloads to `ManifestFile` and `ManifestEntry` to make test cases of reading manifests simpler and shorter. These can be in a separate PR.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
Review Comment:
String comparison would be slow. Can we simply use switch case like below:
```
switch (idx) {
case 0:
// deal with ManifestFile::kManifestPath
case 1:
// deal with ManifestFile::kManifestLength
...
}
```
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,20 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<ManifestEntry>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static Result<std::shared_ptr<ManifestReader>> MakeReader(
Review Comment:
```suggestion
static Result<std::unique_ptr<ManifestReader>> MakeReader(
```
Users can cast to shared_ptr from unique_ptr but cannot from the inverse direction.
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx].manifest_path = path_str;
+ }
+ }
+ } else if (field_name == ManifestFile::kManifestLength.name()) {
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ } else if (field_name == ManifestFile::kPartitionSpecId.name()) {
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ } else if (field_name == ManifestFile::kContent.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ } else if (field_name == ManifestFile::kSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kMinSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kAddedSnapshotId.name()) {
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ } else if (field_name == ManifestFile::kAddedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ } else if (field_name == ManifestFile::kExistingFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ } else if (field_name == ManifestFile::kDeletedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ } else if (field_name == ManifestFile::kAddedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kExistingRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kDeletedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kPartitions.name()) {
Review Comment:
What about adding a separate function for the logic in this branch?
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
Review Comment:
We may need to return error if a required field is null. Same for other places.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2204198285
##########
src/iceberg/arrow_c_data_guard_internal.h:
##########
Review Comment:
Maybe we want to evaluate Nanoarrow eventually? As it has these helpers
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
Review Comment:
Has this been addressed? Nanoarrow uses errno error codes so you can get the message that way
##########
test/manifest_list_reader_test.cc:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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 <arrow/filesystem/localfs.h>
+#include <avro/GenericDatum.hh>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_reader.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_reader.h"
+#include "iceberg/schema.h"
+#include "matchers.h"
+#include "temp_file_test_base.h"
+#include "test_common.h"
+
+namespace iceberg {
+
+class ManifestListReaderTest : public TempFileTestBase {
+ protected:
+ static void SetUpTestSuite() { avro::AvroReader::Register(); }
+
+ void SetUp() override {
+ TempFileTestBase::SetUp();
+ local_fs_ = std::make_shared<::arrow::fs::LocalFileSystem>();
+ file_io_ = std::make_shared<iceberg::arrow::ArrowFileSystemFileIO>(local_fs_);
+ }
+
+ std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;
+ std::shared_ptr<FileIO> file_io_;
+};
+
+TEST_F(ManifestListReaderTest, BasicTest) {
+ std::string path = GetResourcePath(
+ "snap-7412193043800610213-1-2bccd69e-d642-4816-bba0-261cd9bd0d93.avro");
+ auto manifest_reader_result = ManifestListReader::MakeReader(path, file_io_);
+ ASSERT_EQ(manifest_reader_result.has_value(), true);
+ auto manifest_reader = std::move(manifest_reader_result.value());
+ auto read_result = manifest_reader->Files();
+ ASSERT_EQ(read_result.has_value(), true);
+ ASSERT_EQ(read_result.value().size(), 4);
+ std::string test_dir_prefix = "/tmp/db/db/iceberg_test/metadata/";
+ for (const auto& file : read_result.value()) {
+ auto manifest_path = file.manifest_path.substr(test_dir_prefix.size());
+ if (manifest_path == "2bccd69e-d642-4816-bba0-261cd9bd0d93-m0.avro") {
Review Comment:
Maybe we also need to assert that we actually see each of these manifests exactly once?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2203922683
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
Review Comment:
there is no code in ArrowError
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2191959082
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
```suggestion
static std::unique_ptr<ManifestReader> MakeReader(std::unique_ptr<Reader> reader);
```
It is a convention to use Make instead of New, Create, etc.
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
BTW, should we create the reader internally using ReaderFactory? We know the manifest schema and file format. Sometimes it is useful to directly return Arrow arrays of manifest entries. Therefore I propose the following interface and make them all static functions:
```cpp
class ICEBERG_EXPORT ManifestReader {
public:
static Result<std::unique_ptr<Reader>> MakeReader(const std::string& manifest_list_location, FileIO& file_io[, std::shared_ptr<Schema> schema]);
static Result<std::vector<std::unique_ptr<ManifestEntry>>> ReadEntries(const std::string& manifest_list_location, FileIO& file_io);
};
```
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
};
/// \brief Read manifest files from a manifest list file.
class ICEBERG_EXPORT ManifestListReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestFile>>> Files() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestFile>>> Files() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestListReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2192005049
##########
src/iceberg/manifest_reader.h:
##########
@@ -34,19 +34,17 @@ namespace iceberg {
/// \brief Read manifest entries from a manifest file.
class ICEBERG_EXPORT ManifestReader {
public:
- virtual Result<std::span<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
+ virtual Result<std::vector<std::unique_ptr<ManifestEntry>>> Entries() const = 0;
- private:
- std::unique_ptr<Reader> reader_;
+ static std::shared_ptr<ManifestReader> NewReader(std::unique_ptr<Reader> reader);
Review Comment:
For ReadEntries, I also think `std::vector<ManifestEntry>` looks better than `std::vector<std::unique_ptr<ManifestEntry>>`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "gty404 (via GitHub)" <gi...@apache.org>.
gty404 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2206233437
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,251 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define NANOARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Status ParsePartitionFieldSummaryList(ArrowArrayView* view_of_column,
+ std::vector<ManifestFile>& manifest_files) {
+ auto manifest_count = view_of_column->length;
+ // view_of_column is list<struct<PartitionFieldSummary>>
+ if (view_of_column->storage_type != ArrowType::NANOARROW_TYPE_LIST) {
+ return InvalidManifestList("partitions field should be a list.");
+ }
+ auto view_of_list_iterm = view_of_column->children[0];
+ // view_of_list_iterm is struct<PartitionFieldSummary>
+ if (view_of_list_iterm->storage_type != ArrowType::NANOARROW_TYPE_STRUCT) {
+ return InvalidManifestList("partitions list field should be a list.");
+ }
+ if (view_of_list_iterm->n_children != 4) {
+ return InvalidManifestList("PartitionFieldSummary should have 4 fields.");
+ }
+ if (view_of_list_iterm->children[0]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_null should have be bool type column.");
+ }
+ auto contains_null = view_of_list_iterm->children[0];
+ if (view_of_list_iterm->children[1]->storage_type != ArrowType::NANOARROW_TYPE_BOOL) {
+ return InvalidManifestList("contains_nan should have be bool type column.");
+ }
+ auto contains_nan = view_of_list_iterm->children[1];
+ if (view_of_list_iterm->children[2]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("lower_bound should have be binary type column.");
+ }
+ auto lower_bound_list = view_of_list_iterm->children[2];
+ if (view_of_list_iterm->children[3]->storage_type != ArrowType::NANOARROW_TYPE_BINARY) {
+ return InvalidManifestList("upper_bound should have be binary type column.");
+ }
+ auto upper_bound_list = view_of_list_iterm->children[3];
+ for (int64_t manifest_idx = 0; manifest_idx < manifest_count; manifest_idx++) {
+ auto offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx);
+ auto next_offset = ArrowArrayViewListChildOffset(view_of_column, manifest_idx + 1);
+ // partitions from offset to next_offset belongs to manifest_idx
+ auto& manifest_file = manifest_files[manifest_idx];
+ for (int64_t partition_idx = offset; partition_idx < next_offset; partition_idx++) {
+ PartitionFieldSummary partition_field_summary;
+ if (!ArrowArrayViewIsNull(contains_null, partition_idx)) {
+ partition_field_summary.contains_null =
+ ArrowArrayViewGetIntUnsafe(contains_null, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(contains_nan, partition_idx)) {
+ partition_field_summary.contains_nan =
+ ArrowArrayViewGetIntUnsafe(contains_nan, partition_idx);
+ }
+ if (!ArrowArrayViewIsNull(lower_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(lower_bound_list, partition_idx);
+ partition_field_summary.lower_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+ if (!ArrowArrayViewIsNull(upper_bound_list, partition_idx)) {
+ auto buffer = ArrowArrayViewGetBytesUnsafe(upper_bound_list, partition_idx);
+ partition_field_summary.upper_bound = std::vector<uint8_t>(
+ buffer.data.as_char, buffer.data.as_char + buffer.size_bytes);
+ }
+
+ manifest_file.partitions.emplace_back(partition_field_summary);
+ }
+ }
+ return {};
+}
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidManifestList("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ ArrowArrayView array_view;
+ auto status = ArrowArrayViewInitFromSchema(&array_view, schema, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ internal::ArrowArrayViewGuard view_guard(&array_view);
+ status = ArrowArrayViewSetArray(&array_view, array_in, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+ status = ArrowArrayViewValidate(&array_view, NANOARROW_VALIDATION_LEVEL_FULL, &error);
+ NANOARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidSchema("Field index {} is not found in schema", idx);
+ }
+ auto field_name = field.value().get().name();
+ bool required = !field.value().get().optional();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(item, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
Review Comment:
To address the issue of compatibility between versions, I understand that there are two approaches:
1. The io layer is only responsible for executing deserialization accurately without making excessive assumptions. Compatibility is handled by the upper layer, as it has more information to assist in interpreting missing fields, such as the table format version existing in the table metadata.
2. During the underlying deserialization, supplement missing fields to be transparent to the upper layer. This requires that new fields all have reasonable default values, so that the upper layer always sees the latest version of the meta.
Both approaches require that the supplementation of missing values be as converged as possible to the same place, to avoid each logic that consumes the field needing to handle compatibility issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2203979933
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
+ }
+
+Result<std::vector<ManifestFile>> ParseManifestListEntry(ArrowSchema* schema,
+ ArrowArray* array_in,
+ const Schema& iceberg_schema) {
+ if (schema->n_children != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ schema->n_children, array_in->n_children);
+ }
+ if (iceberg_schema.fields().size() != array_in->n_children) {
+ return InvalidArgument("Columns size not match between schema:{} and array:{}",
+ iceberg_schema.fields().size(), array_in->n_children);
+ }
+
+ ArrowError error;
+ 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);
+ ARROW_RETURN_IF_NOT_OK(status, error);
+
+ std::vector<ManifestFile> manifest_files;
+ manifest_files.resize(array_in->length);
+
+ for (int64_t idx = 0; idx < array_in->n_children; idx++) {
+ const auto& field = iceberg_schema.GetFieldByIndex(idx);
+ if (!field.has_value()) {
+ return InvalidArgument("Field not found in schema: {}", idx);
+ }
+ auto field_name = field.value().get().name();
+ auto view_of_column = array_view.children[idx];
+
+#define PARSE_PRIMITIVE_FIELD(field_name, type) \
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) { \
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) { \
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx); \
+ manifest_files[row_idx].field_name = static_cast<type>(value); \
+ } \
+ }
+
+ if (field_name == ManifestFile::kManifestPath.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetStringUnsafe(view_of_column, row_idx);
+ std::string path_str(value.data, value.size_bytes);
+ manifest_files[row_idx].manifest_path = path_str;
+ }
+ }
+ } else if (field_name == ManifestFile::kManifestLength.name()) {
+ PARSE_PRIMITIVE_FIELD(manifest_length, int64_t);
+ } else if (field_name == ManifestFile::kPartitionSpecId.name()) {
+ PARSE_PRIMITIVE_FIELD(partition_spec_id, int32_t);
+ } else if (field_name == ManifestFile::kContent.name()) {
+ for (size_t row_idx = 0; row_idx < view_of_column->length; row_idx++) {
+ if (!ArrowArrayViewIsNull(view_of_column, row_idx)) {
+ auto value = ArrowArrayViewGetIntUnsafe(view_of_column, row_idx);
+ manifest_files[row_idx].content = static_cast<ManifestFile::Content>(value);
+ }
+ }
+ } else if (field_name == ManifestFile::kSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kMinSequenceNumber.name()) {
+ PARSE_PRIMITIVE_FIELD(min_sequence_number, int64_t);
+ } else if (field_name == ManifestFile::kAddedSnapshotId.name()) {
+ PARSE_PRIMITIVE_FIELD(added_snapshot_id, int64_t);
+ } else if (field_name == ManifestFile::kAddedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_files_count, int32_t);
+ } else if (field_name == ManifestFile::kExistingFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_files_count, int32_t);
+ } else if (field_name == ManifestFile::kDeletedFilesCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_files_count, int32_t);
+ } else if (field_name == ManifestFile::kAddedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(added_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kExistingRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(existing_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kDeletedRowsCount.name()) {
+ PARSE_PRIMITIVE_FIELD(deleted_rows_count, int64_t);
+ } else if (field_name == ManifestFile::kPartitions.name()) {
Review Comment:
logical to parse std::vector<PartitionFieldSummary> partitions is really, i will make a new function for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2203940400
##########
src/iceberg/manifest_reader_internal.cc:
##########
@@ -0,0 +1,233 @@
+/*
+ * 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 "manifest_reader_internal.h"
+
+#include <array>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow_c_data_guard_internal.h"
+#include "iceberg/manifest_entry.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/type.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+#define ARROW_RETURN_IF_NOT_OK(status, error) \
+ if (status != NANOARROW_OK) [[unlikely]] { \
+ return InvalidArrowData("Nanoarrow error: {}", error.message); \
Review Comment:
I mean `status` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add manifest list reader [iceberg-cpp]
Posted by "dongxiao1198 (via GitHub)" <gi...@apache.org>.
dongxiao1198 commented on code in PR #143:
URL: https://github.com/apache/iceberg-cpp/pull/143#discussion_r2204586061
##########
test/manifest_list_reader_test.cc:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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 <arrow/filesystem/localfs.h>
+#include <avro/GenericDatum.hh>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/avro/avro_reader.h"
+#include "iceberg/manifest_list.h"
+#include "iceberg/manifest_reader.h"
+#include "iceberg/schema.h"
+#include "matchers.h"
+#include "temp_file_test_base.h"
+#include "test_common.h"
+
+namespace iceberg {
+
+class ManifestListReaderTest : public TempFileTestBase {
+ protected:
+ static void SetUpTestSuite() { avro::AvroReader::Register(); }
+
+ void SetUp() override {
+ TempFileTestBase::SetUp();
+ local_fs_ = std::make_shared<::arrow::fs::LocalFileSystem>();
+ file_io_ = std::make_shared<iceberg::arrow::ArrowFileSystemFileIO>(local_fs_);
+ }
+
+ std::shared_ptr<::arrow::fs::LocalFileSystem> local_fs_;
+ std::shared_ptr<FileIO> file_io_;
+};
+
+TEST_F(ManifestListReaderTest, BasicTest) {
+ std::string path = GetResourcePath(
+ "snap-7412193043800610213-1-2bccd69e-d642-4816-bba0-261cd9bd0d93.avro");
+ auto manifest_reader_result = ManifestListReader::MakeReader(path, file_io_);
+ ASSERT_EQ(manifest_reader_result.has_value(), true);
+ auto manifest_reader = std::move(manifest_reader_result.value());
+ auto read_result = manifest_reader->Files();
+ ASSERT_EQ(read_result.has_value(), true);
+ ASSERT_EQ(read_result.value().size(), 4);
+ std::string test_dir_prefix = "/tmp/db/db/iceberg_test/metadata/";
+ for (const auto& file : read_result.value()) {
+ auto manifest_path = file.manifest_path.substr(test_dir_prefix.size());
+ if (manifest_path == "2bccd69e-d642-4816-bba0-261cd9bd0d93-m0.avro") {
Review Comment:
I will implement the operator== and validator of manifest list in my next pr to check the file
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org