wgtmac commented on code in PR #714:
URL: https://github.com/apache/iceberg-cpp/pull/714#discussion_r3831443300
##########
src/iceberg/parquet/parquet_reader.cc:
##########
@@ -250,6 +322,24 @@ class ParquetReader::Impl {
reader_->GetRecordBatchReader(row_group_indices, column_indices));
}
+ // Build the output Arrow schema from the projected Iceberg schema. This
schema is the
+ // target of ProjectRecordBatch, so it must describe the projected schema
rather than
+ // the schema of the file.
+ ArrowSchema arrow_schema;
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema));
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_,
+ ::arrow::ImportSchema(&arrow_schema));
+
+ if (use_large_list_ && ProducesLargeList(*context_->record_batch_reader_))
{
+ // Align the output schema with the large_list arrays produced by the
Parquet
+ // reader. Note that Arrow ignores the requested list type when the file
carries
+ // serialized ARROW:schema metadata, in which case the reader keeps
producing plain
+ // list arrays and the output schema must keep describing them as such.
+ context_->output_arrow_schema_ =
+
::arrow::schema(UseLargeListFields(context_->output_arrow_schema_->fields()),
Review Comment:
This rewrite is global: one `large_list` makes every projected list a
`large_list`. Stored Arrow schemas can contain both types, so please map each
projected field recursively from the corresponding reader field instead of
using `ProducesLargeList()` as a single flag. For missing or default fields,
use the configured output preference.
##########
src/iceberg/parquet/parquet_reader.cc:
##########
@@ -250,6 +322,24 @@ class ParquetReader::Impl {
reader_->GetRecordBatchReader(row_group_indices, column_indices));
}
+ // Build the output Arrow schema from the projected Iceberg schema. This
schema is the
+ // target of ProjectRecordBatch, so it must describe the projected schema
rather than
+ // the schema of the file.
+ ArrowSchema arrow_schema;
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema));
+ ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_,
+ ::arrow::ImportSchema(&arrow_schema));
+
+ if (use_large_list_ && ProducesLargeList(*context_->record_batch_reader_))
{
Review Comment:
Please align the output schema with the reader schema even when this option
is false. Arrow restores a stored `large_list` regardless of `set_list_type`,
so this path currently leaves the output as `list` and casts a `LargeListArray`
as `ListArray`.
##########
src/iceberg/test/parquet_test.cc:
##########
@@ -267,6 +293,38 @@ class ParquetReaderTest : public TempFileTestBase {
.data_sequence_number =
data_sequence_number});
}
+ // Writes a list parquet file through parquet::arrow::WriteTable, which
serializes the
+ // Arrow schema of the table into the ARROW:schema key value metadata of the
file.
+ void CreateListParquetFileWithArrowSchema() {
+ const std::string kParquetFieldIdKey = "PARQUET:field_id";
+ auto arrow_schema = ::arrow::schema(
+ {::arrow::field("id", ::arrow::int32(), /*nullable=*/false,
+ ::arrow::KeyValueMetadata::Make({kParquetFieldIdKey},
{"1"})),
+ ::arrow::field(
+ "numbers",
+ ::arrow::list(::arrow::field(
+ "element", ::arrow::int32(), /*nullable=*/true,
+ ::arrow::KeyValueMetadata::Make({kParquetFieldIdKey},
{"101"}))),
+ /*nullable=*/true,
+ ::arrow::KeyValueMetadata::Make({kParquetFieldIdKey}, {"2"}))});
+ auto batch =
+ ::arrow::RecordBatch::FromStructArray(
+
::arrow::json::ArrayFromJSONString(::arrow::struct_(arrow_schema->fields()),
+ R"([[1, [1, 2]], [2, [3]]])")
+ .ValueOrDie())
+ .ValueOrDie();
+ auto table = ::arrow::Table::FromRecordBatches(arrow_schema,
{batch}).ValueOrDie();
+
+ auto io = internal::checked_cast<arrow::ArrowFileSystemFileIO&>(*file_io_);
+ auto outfile = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie();
+
+ // write a single row group so that one batch holds every row
+ ASSERT_TRUE(::parquet::arrow::WriteTable(*table,
::arrow::default_memory_pool(),
Review Comment:
`WriteTable` does not store `ARROW:schema` by default. Please pass
`ArrowWriterProperties::Builder().store_schema()->build()`, assert that the
footer key exists, and cover a stored `large_list` with the default reader
option plus a mixed `list`/`large_list` 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
