You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "wiedld (via GitHub)" <gi...@apache.org> on 2024/12/21 01:07:31 UTC
[PR] ParquetSink should be aware of arrow schema encoding (configurable) in the file metadata. [datafusion]
wiedld opened a new pull request, #13866:
URL: https://github.com/apache/datafusion/pull/13866
## Which issue does this PR close?
Closes https://github.com/apache/datafusion/issues/11770
## Rationale for this change
The [ArrowWriter](https://docs.rs/parquet/53.3.0/parquet/arrow/arrow_writer/struct.ArrowWriter.html) with it's default `ArrowWriterOptions` will encode the arrow schema in the parquet kv_metadata, unless explicitly skipped. Skipping is done via [ArrowWriterOptions::with_skip_arrow_metadata](https://docs.rs/parquet/53.3.0/parquet/arrow/arrow_writer/struct.ArrowWriterOptions.html#method.with_skip_arrow_metadata).
In datafusion's ParquetSink, we can write in either single threaded or parallelized format. When in single-threaded mode, we use the default `ArrowWriterOptions` and the arrow schema is inserted into file kv_meta. However, when performing parallelized writes we do not use the ArrowWriter and instead rely upon the [SerializedFileWriter](https://docs.rs/parquet/53.3.0/parquet/file/writer/struct.SerializedFileWriter.html). As a result, we are missing the arrow schema metadata in the parquet files (see the issue ticket).
### ArrowWriterOptions vs WriterProperties
The SerializedFileWriter, along with other associated writers, rely upon the [WriterProperties](https://docs.rs/parquet/53.3.0/parquet/file/properties/struct.WriterProperties.html). The `WriterProperties` differ from the `ArrowWriterOptions` only in terms of the `skip_arrow_metadata` (the missing configuration cause our current issue), and the `schema_root`:
```
pub struct ArrowWriterOptions {
properties: WriterProperties,
skip_arrow_metadata: bool,
schema_root: Option<String>,
}
```
The `skip_arrow_metadata` is only used to modify the kv_metadata within the WriterProperties. Therefore I focused on updating our DF methods to construct WriterProperties to include this arrow schema (when configured). In this way, we can continue using WriterProperties -- with the added features we are missing from ArrowWriterOptions.
## What changes are included in this PR?
* add a new configuration `ParquetOptions.skip_arrow_metadata`
* have ParquetSink single-threaded writes, which use the ArrowWriter, respect this configuration
* have ParquetSink multiple-threaded writes, which use WriterProperties, respect this configuration
* this is done by considering the inclusion of the arrow schema, or not, during the WriterProperties construction
## Are these changes tested?
yes.
## Are there any user-facing changes?
We have new APIs:
* `ParquetOptions.skip_arrow_metadata` configuration
* replace/deprecate `ParquetWriterOptions::try_from(ParquetWriterOptions)` and replace with methods which explicitly handle arrow 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] WIP: ParquetSink should be aware of arrow schema encoding for the file metadata. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898052582
##########
datafusion/sqllogictest/test_files/repartition_scan.slt:
##########
@@ -61,7 +61,7 @@ logical_plan
physical_plan
01)CoalesceBatchesExec: target_batch_size=8192
02)--FilterExec: column1@0 != 42
-03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..88], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:88..176], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:176..264], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:264..351]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=column1_null_count@2 != column1_row_count@3 AND (column1_min@0 != 42 OR 42 != column1_max@1), required_guarantees=[column1 not in (42)]
+03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..137], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:137..274], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:274..411], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:411..547]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=column1_null_count@2 != column1_row_count@3 AND (column1_min@0 != 42 OR 42 != column1_max@1), required_guarantees=[column1 not in (42)]
Review Comment:
In this file, the partition offsets are shifted based upon the larger kv metadata (from the default addition of encoded arrow 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898709356
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
+ #[deprecated(
+ since = "44.0.0",
+ note = "Please use `TableParquetOptions::into_writer_properties_builder` and `TableParquetOptions::into_writer_properties_builder_with_arrow_schema`"
+ )]
+ pub fn try_from(table_opts: &TableParquetOptions) -> Result<ParquetWriterOptions> {
Review Comment:
outdated. changed approach.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] ParquetSink should be aware of arrow schema encoding (configurable) in the file metadata. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1894526449
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -2335,42 +2347,64 @@ mod tests {
async fn parquet_sink_write() -> Result<()> {
let parquet_sink = create_written_parquet_sink("file:///").await?;
- // assert written
- let mut written = parquet_sink.written();
- let written = written.drain();
- assert_eq!(
- written.len(),
- 1,
- "expected a single parquet files to be written, instead found {}",
- written.len()
- );
+ // assert written to proper path
+ let (path, file_metadata) = get_written(parquet_sink)?;
+ let path_parts = path.parts().collect::<Vec<_>>();
+ assert_eq!(path_parts.len(), 1, "should not have path prefix");
// check the file metadata
- let (
- path,
- FileMetaData {
- num_rows,
- schema,
- key_value_metadata,
- ..
+ let expected_kv_meta = vec![
+ KeyValue {
+ key: "my-data".to_string(),
+ value: Some("stuff".to_string()),
},
- ) = written.take(1).next().unwrap();
+ KeyValue {
+ key: "my-data-bool-key".to_string(),
+ value: None,
+ },
+ ];
+ assert_file_metadata(file_metadata, expected_kv_meta);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn parquet_sink_parallel_write() -> Result<()> {
Review Comment:
New test added.
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -2335,42 +2347,64 @@ mod tests {
async fn parquet_sink_write() -> Result<()> {
let parquet_sink = create_written_parquet_sink("file:///").await?;
- // assert written
- let mut written = parquet_sink.written();
- let written = written.drain();
- assert_eq!(
- written.len(),
- 1,
- "expected a single parquet files to be written, instead found {}",
- written.len()
- );
+ // assert written to proper path
+ let (path, file_metadata) = get_written(parquet_sink)?;
+ let path_parts = path.parts().collect::<Vec<_>>();
+ assert_eq!(path_parts.len(), 1, "should not have path prefix");
// check the file metadata
- let (
- path,
- FileMetaData {
- num_rows,
- schema,
- key_value_metadata,
- ..
+ let expected_kv_meta = vec![
+ KeyValue {
+ key: "my-data".to_string(),
+ value: Some("stuff".to_string()),
},
- ) = written.take(1).next().unwrap();
+ KeyValue {
+ key: "my-data-bool-key".to_string(),
+ value: None,
+ },
+ ];
+ assert_file_metadata(file_metadata, expected_kv_meta);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn parquet_sink_parallel_write() -> Result<()> {
+ let opts = ParquetOptions {
+ allow_single_file_parallelism: true,
+ maximum_parallel_row_group_writers: 2,
+ maximum_buffered_record_batches_per_stream: 2,
+ ..Default::default()
+ };
+
+ let parquet_sink =
+ create_written_parquet_sink_using_config("file:///", opts).await?;
+
+ // assert written to proper path
+ let (path, file_metadata) = get_written(parquet_sink)?;
let path_parts = path.parts().collect::<Vec<_>>();
assert_eq!(path_parts.len(), 1, "should not have path prefix");
- assert_eq!(num_rows, 2, "file metadata to have 2 rows");
- assert!(
- schema.iter().any(|col_schema| col_schema.name == "a"),
- "output file metadata should contain col a"
- );
- assert!(
- schema.iter().any(|col_schema| col_schema.name == "b"),
- "output file metadata should contain col b"
- );
+ // check the file metadata
+ let expected_kv_meta = vec![
+ KeyValue {
+ key: "my-data".to_string(),
+ value: Some("stuff".to_string()),
+ },
+ KeyValue {
+ key: "my-data-bool-key".to_string(),
+ value: None,
+ },
+ ];
+ assert_file_metadata(file_metadata, expected_kv_meta);
- let mut key_value_metadata = key_value_metadata.unwrap();
- key_value_metadata.sort_by(|a, b| a.key.cmp(&b.key));
- let expected_metadata = vec![
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn parquet_sink_write_insert_schema_into_metadata() -> Result<()> {
Review Comment:
New test added.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1899464378
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -750,6 +749,28 @@ impl ParquetSink {
}
}
+ /// Create writer properties based upon configuration settings,
+ /// including partitioning and the inclusion of arrow schema metadata.
+ fn create_writer_props(&self) -> Result<WriterProperties> {
+ let schema = if self.parquet_options.global.allow_single_file_parallelism {
+ // If parallelizing writes, we may be also be doing hive style partitioning
+ // into multiple files which impacts the schema per file.
+ // Refer to `self.get_writer_schema()`
+ &self.get_writer_schema()
+ } else {
+ self.config.output_schema()
+ };
+
+ // TODO: avoid this clone in follow up PR, where the writer properties & schema
Review Comment:
Is this a PR you plan to do ? Should I file a ticket to track?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898612752
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
+ #[deprecated(
+ since = "44.0.0",
+ note = "Please use `TableParquetOptions::into_writer_properties_builder` and `TableParquetOptions::into_writer_properties_builder_with_arrow_schema`"
+ )]
+ pub fn try_from(table_opts: &TableParquetOptions) -> Result<ParquetWriterOptions> {
// ParquetWriterOptions will have defaults for the remaining fields (e.g. sorting_columns)
Ok(ParquetWriterOptions {
- writer_options: WriterPropertiesBuilder::try_from(parquet_table_options)?
- .build(),
+ writer_options: table_opts.into_writer_properties_builder()?.build(),
})
}
-}
-
-impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
- type Error = DataFusionError;
/// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
///
/// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column.
- fn try_from(table_parquet_options: &TableParquetOptions) -> Result<Self> {
+ pub fn into_writer_properties_builder(&self) -> Result<WriterPropertiesBuilder> {
+ self.into_writer_properties_builder_with_arrow_schema(None)
+ }
+
+ /// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
+ ///
+ /// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column,
+ /// as well as the arrow schema encoded into the kv_meta at [`ARROW_SCHEMA_META_KEY`].
+ pub fn into_writer_properties_builder_with_arrow_schema(
Review Comment:
> So for example I would expect that ParquetOptions::into_writer_properties_builder would always set the arrow metadata to be consistent with how WriterProperties works 🤔
The [ParquetOptions](https://docs.rs/datafusion/latest/datafusion/config/struct.ParquetOptions.html) does not have kv_metadata. That is held within the parent structure [TableParquetOptions](https://docs.rs/datafusion/latest/datafusion/common/config/struct.TableParquetOptions.html). That is why the existing [TryFrom code only constructs the writer properties from the TableParquetOptions](https://github.com/apache/datafusion/blob/a08dc0af8f798afe73a2fdf170ab36e72ad7e782/datafusion/common/src/file_options/parquet_writer.rs#L66). Not the ParquetOptions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898714075
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -140,6 +162,32 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
}
}
+/// Encodes the Arrow schema into the IPC format, and base64 encodes it
+///
+/// TODO: make arrow schema encoding available in a public API.
+/// Refer to currently private `add_encoded_arrow_schema_to_metadata` and `encode_arrow_schema` public.
+/// <https://github.com/apache/arrow-rs/blob/2908a80d9ca3e3fb0414e35b67856f1fb761304c/parquet/src/arrow/schema/mod.rs#L172-L221>
+fn encode_arrow_schema(schema: &Arc<Schema>) -> String {
Review Comment:
https://github.com/apache/arrow-rs/pull/6916
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] WIP: ParquetSink should be aware of arrow schema encoding for the file metadata. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898052582
##########
datafusion/sqllogictest/test_files/repartition_scan.slt:
##########
@@ -61,7 +61,7 @@ logical_plan
physical_plan
01)CoalesceBatchesExec: target_batch_size=8192
02)--FilterExec: column1@0 != 42
-03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..88], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:88..176], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:176..264], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:264..351]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=column1_null_count@2 != column1_row_count@3 AND (column1_min@0 != 42 OR 42 != column1_max@1), required_guarantees=[column1 not in (42)]
+03)----ParquetExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:0..137], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:137..274], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:274..411], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/parquet_table/2.parquet:411..547]]}, projection=[column1], predicate=column1@0 != 42, pruning_predicate=column1_null_count@2 != column1_row_count@3 AND (column1_min@0 != 42 OR 42 != column1_max@1), required_guarantees=[column1 not in (42)]
Review Comment:
In this file, the partition offsets are shifted based upon the slightly larger kv metadata (from the default addition of encoded arrow 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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #13866:
URL: https://github.com/apache/datafusion/pull/13866
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898604502
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -2519,6 +2612,42 @@ mod tests {
Ok(parquet_sink)
}
+ fn get_written(parquet_sink: Arc<ParquetSink>) -> Result<(Path, FileMetaData)> {
+ let mut written = parquet_sink.written();
+ let written = written.drain();
+ assert_eq!(
+ written.len(),
+ 1,
+ "expected a single parquet files to be written, instead found {}",
+ written.len()
+ );
+
+ let (path, file_metadata) = written.take(1).next().unwrap();
+ Ok((path, file_metadata))
+ }
+
+ fn assert_file_metadata(file_metadata: FileMetaData, expected_kv: Vec<KeyValue>) {
Review Comment:
The second arg (`expected_kv`) as a ref does indeed avoid the cloning. Thank you.
The first arg does not have any cloning currently (see tests), and if we add the ref -- then I would need to clone the `file_metadata.key_value_metadata` in order to sort & compare to the `expected_kv`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898655480
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
+ #[deprecated(
+ since = "44.0.0",
+ note = "Please use `TableParquetOptions::into_writer_properties_builder` and `TableParquetOptions::into_writer_properties_builder_with_arrow_schema`"
+ )]
+ pub fn try_from(table_opts: &TableParquetOptions) -> Result<ParquetWriterOptions> {
// ParquetWriterOptions will have defaults for the remaining fields (e.g. sorting_columns)
Ok(ParquetWriterOptions {
- writer_options: WriterPropertiesBuilder::try_from(parquet_table_options)?
- .build(),
+ writer_options: table_opts.into_writer_properties_builder()?.build(),
})
}
-}
-
-impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
- type Error = DataFusionError;
/// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
///
/// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column.
- fn try_from(table_parquet_options: &TableParquetOptions) -> Result<Self> {
+ pub fn into_writer_properties_builder(&self) -> Result<WriterPropertiesBuilder> {
+ self.into_writer_properties_builder_with_arrow_schema(None)
+ }
+
+ /// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
+ ///
+ /// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column,
+ /// as well as the arrow schema encoded into the kv_meta at [`ARROW_SCHEMA_META_KEY`].
+ pub fn into_writer_properties_builder_with_arrow_schema(
Review Comment:
Here is the change: https://github.com/apache/datafusion/pull/13866/commits/f2f9b00abd9e783fdbc32e8b167251423f483fbc
I also made another change to prevent the user errors I was worried about: https://github.com/apache/datafusion/pull/13866/commits/c5ad794bb9652e84e9ad7a66f15bef964fb2a4dc
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#issuecomment-2565373933
I'll plan to merge this tomorrow unless others would like time to review
Thanks again @wiedld
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#issuecomment-2567001233
Thanks again @wiedld
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898612752
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
+ #[deprecated(
+ since = "44.0.0",
+ note = "Please use `TableParquetOptions::into_writer_properties_builder` and `TableParquetOptions::into_writer_properties_builder_with_arrow_schema`"
+ )]
+ pub fn try_from(table_opts: &TableParquetOptions) -> Result<ParquetWriterOptions> {
// ParquetWriterOptions will have defaults for the remaining fields (e.g. sorting_columns)
Ok(ParquetWriterOptions {
- writer_options: WriterPropertiesBuilder::try_from(parquet_table_options)?
- .build(),
+ writer_options: table_opts.into_writer_properties_builder()?.build(),
})
}
-}
-
-impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
- type Error = DataFusionError;
/// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
///
/// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column.
- fn try_from(table_parquet_options: &TableParquetOptions) -> Result<Self> {
+ pub fn into_writer_properties_builder(&self) -> Result<WriterPropertiesBuilder> {
+ self.into_writer_properties_builder_with_arrow_schema(None)
+ }
+
+ /// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
+ ///
+ /// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column,
+ /// as well as the arrow schema encoded into the kv_meta at [`ARROW_SCHEMA_META_KEY`].
+ pub fn into_writer_properties_builder_with_arrow_schema(
Review Comment:
> So for example I would expect that ParquetOptions::into_writer_properties_builder would always set the arrow metadata to be consistent with how WriterProperties works 🤔
The [ParquetOptions](https://docs.rs/datafusion/latest/datafusion/config/struct.ParquetOptions.html) does not have kv_metadata. That is held within the parent structure [TableParquetOptions](https://docs.rs/datafusion/latest/datafusion/common/config/struct.TableParquetOptions.html). That is why the existing [TryFrom code only constructs the writer properties from the TableParquetOptions](https://github.com/apache/datafusion/blob/a08dc0af8f798afe73a2fdf170ab36e72ad7e782/datafusion/common/src/file_options/parquet_writer.rs#L66). Not the ParquetOptions.
I like the idea of making the metadata APIs on the `TableParquetOptions`, altho I'll keep the config on the ParquetOptions. I'll push up the change shortly.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] ParquetSink should be aware of arrow schema encoding (configurable) in the file metadata. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1894525850
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
+ #[deprecated(
+ since = "44.0.0",
+ note = "Please use `TableParquetOptions::into_writer_properties_builder` and `TableParquetOptions::into_writer_properties_builder_with_arrow_schema`"
+ )]
+ pub fn try_from(table_opts: &TableParquetOptions) -> Result<ParquetWriterOptions> {
Review Comment:
I cannot deprecate trait impls. Instead, this at least gives a deprecation notice under some conditions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Supporting writing schema metadata when writing Parquet in parallel [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898599827
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -750,6 +749,26 @@ impl ParquetSink {
}
}
+ /// Create writer properties based upon configuration settings,
+ /// including partitioning and the inclusion of arrow schema metadata.
+ fn create_writer_props(&self) -> Result<WriterProperties> {
+ let props = if !self.parquet_options.global.skip_arrow_metadata {
+ let schema = if self.parquet_options.global.allow_single_file_parallelism {
Review Comment:
This is the potential for hive-style partitioning (across multiple files) which impacts the schema per file. I've added a code comment to explain.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] ParquetSink should be aware of arrow schema encoding for the file metadata. [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1898124585
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
Review Comment:
It would be nice to avoid having to change this API -- I have a suggestion below
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -51,38 +58,53 @@ impl ParquetWriterOptions {
}
}
-impl TryFrom<&TableParquetOptions> for ParquetWriterOptions {
- type Error = DataFusionError;
-
- fn try_from(parquet_table_options: &TableParquetOptions) -> Result<Self> {
+impl TableParquetOptions {
+ #[deprecated(
+ since = "44.0.0",
+ note = "Please use `TableParquetOptions::into_writer_properties_builder` and `TableParquetOptions::into_writer_properties_builder_with_arrow_schema`"
+ )]
+ pub fn try_from(table_opts: &TableParquetOptions) -> Result<ParquetWriterOptions> {
// ParquetWriterOptions will have defaults for the remaining fields (e.g. sorting_columns)
Ok(ParquetWriterOptions {
- writer_options: WriterPropertiesBuilder::try_from(parquet_table_options)?
- .build(),
+ writer_options: table_opts.into_writer_properties_builder()?.build(),
})
}
-}
-
-impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
- type Error = DataFusionError;
/// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
///
/// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column.
- fn try_from(table_parquet_options: &TableParquetOptions) -> Result<Self> {
+ pub fn into_writer_properties_builder(&self) -> Result<WriterPropertiesBuilder> {
+ self.into_writer_properties_builder_with_arrow_schema(None)
+ }
+
+ /// Convert the session's [`TableParquetOptions`] into a single write action's [`WriterPropertiesBuilder`].
+ ///
+ /// The returned [`WriterPropertiesBuilder`] includes customizations applicable per column,
+ /// as well as the arrow schema encoded into the kv_meta at [`ARROW_SCHEMA_META_KEY`].
+ pub fn into_writer_properties_builder_with_arrow_schema(
Review Comment:
I think making setting the arrow schema a different function might be more consistent with the rest of this API https://docs.rs/datafusion/latest/datafusion/config/struct.ParquetOptions.html
So for example I would expect that `ParquetOptions::into_writer_properties_builder` would always set the arrow metadata to be consistent with how `WriterProperties` works 🤔
If someone doesn't want the arrow metadata, I would expect an option to disable it, like https://docs.rs/parquet/53.3.0/parquet/arrow/arrow_writer/struct.ArrowWriterOptions.html#method.with_skip_arrow_metadata
So that might mean that `TableParquetOptions` has a field like `skip_arrow_metadata`
And then depending on the value of that field, `into_writer_properties_builder` would set/not set the metadata appropriately
That would also avoid having to change the conversion API
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -2519,6 +2612,42 @@ mod tests {
Ok(parquet_sink)
}
+ fn get_written(parquet_sink: Arc<ParquetSink>) -> Result<(Path, FileMetaData)> {
+ let mut written = parquet_sink.written();
+ let written = written.drain();
+ assert_eq!(
+ written.len(),
+ 1,
+ "expected a single parquet files to be written, instead found {}",
+ written.len()
+ );
+
+ let (path, file_metadata) = written.take(1).next().unwrap();
+ Ok((path, file_metadata))
+ }
+
+ fn assert_file_metadata(file_metadata: FileMetaData, expected_kv: Vec<KeyValue>) {
Review Comment:
If you passed in a ref here it might make the code above simpler (less clone):
```suggestion
fn assert_file_metadata(file_metadata: &FileMetaData, expected_kv: &[KeyValue]) {
```
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -759,10 +778,14 @@ impl ParquetSink {
parquet_props: WriterProperties,
) -> Result<AsyncArrowWriter<BufWriter>> {
let buf_writer = BufWriter::new(object_store, location.clone());
- let writer = AsyncArrowWriter::try_new(
+ let options = ArrowWriterOptions::new()
Review Comment:
this is nice
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -750,6 +749,26 @@ impl ParquetSink {
}
}
+ /// Create writer properties based upon configuration settings,
+ /// including partitioning and the inclusion of arrow schema metadata.
+ fn create_writer_props(&self) -> Result<WriterProperties> {
+ let props = if !self.parquet_options.global.skip_arrow_metadata {
+ let schema = if self.parquet_options.global.allow_single_file_parallelism {
Review Comment:
Can you explain why different schemas are needed for single file parallelism?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] ParquetSink should be aware of arrow schema encoding (configurable) in the file metadata. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1894526384
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -2335,42 +2347,64 @@ mod tests {
async fn parquet_sink_write() -> Result<()> {
let parquet_sink = create_written_parquet_sink("file:///").await?;
- // assert written
- let mut written = parquet_sink.written();
- let written = written.drain();
- assert_eq!(
- written.len(),
- 1,
- "expected a single parquet files to be written, instead found {}",
- written.len()
- );
+ // assert written to proper path
+ let (path, file_metadata) = get_written(parquet_sink)?;
+ let path_parts = path.parts().collect::<Vec<_>>();
+ assert_eq!(path_parts.len(), 1, "should not have path prefix");
Review Comment:
I refactored the tests first, into a series of helpers. May be easier to review in this commit: https://github.com/apache/datafusion/pull/13866/commits/09004c56e6e4b96d3767e03a81f3eb228b3529f1
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] ParquetSink should be aware of arrow schema encoding (configurable) in the file metadata. [datafusion]
Posted by "wiedld (via GitHub)" <gi...@apache.org>.
wiedld commented on code in PR #13866:
URL: https://github.com/apache/datafusion/pull/13866#discussion_r1894526006
##########
datafusion/common/src/file_options/parquet_writer.rs:
##########
@@ -140,6 +162,32 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
}
}
+/// Encodes the Arrow schema into the IPC format, and base64 encodes it
+///
+/// TODO: make arrow schema encoding available in a public API.
+/// Refer to currently private `add_encoded_arrow_schema_to_metadata` and `encode_arrow_schema` public.
+/// <https://github.com/apache/arrow-rs/blob/2908a80d9ca3e3fb0414e35b67856f1fb761304c/parquet/src/arrow/schema/mod.rs#L172-L221>
+fn encode_arrow_schema(schema: &Arc<Schema>) -> String {
Review Comment:
If we are in agreement on need, I'll go make the arrow-rs PR.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org