You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "ZENOTME (via GitHub)" <gi...@apache.org> on 2024/01/29 06:55:00 UTC
[PR] feat: add parquet writer [iceberg-rust]
ZENOTME opened a new pull request, #176:
URL: https://github.com/apache/iceberg-rust/pull/176
This PR adds a parquet writer under file writer.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470721802
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
> I find that the `AsyncArrowWriter` already provides the [write with a buffer](https://docs.rs/parquet/latest/parquet/arrow/async_writer/struct.AsyncArrowWriter.html#method.write). In this case, maybe we can use this buffer.
Thanks for pointing out! Seems `AsyncArrowWriter` has it own internal buffer, so we don't need to double buffer it anymore.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470672780
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
> There is another thing we can do is adding concurrent here so we can upload parts concurrently. But we need to make a design on how users control the concurrency. Maybe we can discuss them later.
Yes, we can open a discussion about it. IMO, the concurrent writer is used in the case that writes multiple files concurrently. So concurrent writer may should be a higher-level writer.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492445165
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+use crate::spec::DataFileFormat;
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
Review Comment:
The `LocationGenerator` is to generate the path for data dir, such as `/table/data` or `/table/id=1/data/`. `FileNameGenerator` generates the file name, such as `1-1.parquet` or `1-1-delete.parquet`. The benefit of separating them is that we can let users reuse one part of it. E.g. default `LocationGenerator` and custom `FileNameGenerator`. And that's iceberg-java support. In iceberg-java, [`LocationProvider` ](https://github.com/apache/iceberg/blob/062704283918ae843d0e40b3f279f1482f0835e5/api/src/main/java/org/apache/iceberg/io/LocationProvider.java#L30)is to provide the data dir. [The ` FileNameGenerator` is in OutFileFactory.](https://github.com/apache/iceberg/blob/062704283918ae843d0e40b3f279f1482f0835e5/core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java#L93)
--
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 parquet writer [iceberg-rust]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1507992831
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,385 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
+ )?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = max(Self::MIN_BUFFER_SIZE, self.init_buffer_size);
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to build parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+ field_ids: Vec<i32>,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ field_ids: &[i32],
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> Result<DataFileBuilder> {
+ // Only enter here when the file is not empty.
+ assert!(!metadata.row_groups.is_empty());
+ if field_ids.len() != metadata.row_groups[0].columns.len() {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ "Len of field id is not match with len of columns in parquet metadata.",
+ ));
+ }
+
+ let (column_sizes, value_counts, null_value_counts) =
+ {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group.columns.iter().zip(field_ids.iter()).for_each(
+ |(column_chunk, &field_id)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num.entry(field_id).or_insert(0_u64) +=
+ column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0) as u64;
+ }
+ },
+ )
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_path(file_path)
+ .file_format(DataFileFormat::Parquet)
+ .record_count(metadata.num_rows as u64)
+ .file_size_in_bytes(written_size as u64)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ // # TODO
+ // - nan_value_counts
+ // - lower_bounds
+ // - upper_bounds
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ Ok(builder)
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to write using parquet writer.",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to close parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
+
+ Ok(vec![Self::to_data_file_builder(
+ &self.field_ids,
+ metadata,
+ written_size as usize,
+ self.out_file.location().to_string(),
+ )?])
+ }
+}
+
+impl CurrentFileStatus for ParquetWriter {
+ fn current_file_path(&self) -> String {
+ self.out_file.location().to_string()
+ }
+
+ fn current_row_num(&self) -> usize {
+ self.current_row_num
+ }
+
+ fn current_written_size(&self) -> usize {
+ self.written_size.load(std::sync::atomic::Ordering::Relaxed) as usize
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use anyhow::Result;
+ use arrow_array::ArrayRef;
+ use arrow_array::Int64Array;
+ use arrow_array::RecordBatch;
+ use bytes::Bytes;
+ use futures::AsyncReadExt;
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+ use tempfile::TempDir;
+
+ use super::*;
+ use crate::io::FileIOBuilder;
+ use crate::spec::Struct;
+ use crate::writer::file_writer::location_generator::test::MockLocationGenerator;
+ use crate::writer::file_writer::location_generator::DefaultFileNameGenerator;
+
+ #[derive(Clone)]
+ struct TestLocationGen;
+
+ #[tokio::test]
+ async fn test_parquet_writer() -> Result<()> {
+ let temp_dir = TempDir::new().unwrap();
+ let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+ let loccation_gen =
+ MockLocationGenerator::new(temp_dir.path().to_str().unwrap().to_string());
+ let file_name_gen =
+ DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet);
+
+ // prepare data
+ let schema = {
Review Comment:
Can we add some complex types here as well, to assert how the schema is constructed and written (with the focus on the field-IDs)
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+use crate::spec::DataFileFormat;
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate an absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
+}
+
+/// `FileNameGeneratorTrait` used to generate file name for data file. The file name can be passed to `LocationGenerator` to generate the location of the file.
+pub trait FileNameGenerator: Clone + Send + 'static {
+ /// Generate a file name.
+ fn generate_file_name(&self) -> String;
+}
+
+/// `DefaultFileNameGenerator` used to generate file name for data file. The file name can be
+/// passed to `LocationGenerator` to generate the location of the file.
+/// The file name format is "{prefix}-{file_count}[-{suffix}].{file_format}".
+#[derive(Clone)]
+pub struct DefaultFileNameGenerator {
+ prefix: String,
+ suffix: String,
+ format: String,
+ file_count: Arc<AtomicU64>,
+}
+
+impl DefaultFileNameGenerator {
+ /// Create a new `FileNameGenerator`.
+ pub fn new(prefix: String, suffix: Option<String>, format: DataFileFormat) -> Self {
+ let suffix = if let Some(suffix) = suffix {
+ format!("-{}", suffix)
+ } else {
+ "".to_string()
+ };
+
+ Self {
+ prefix,
+ suffix,
+ format: format.to_string(),
+ file_count: Arc::new(AtomicU64::new(0)),
+ }
+ }
+}
+
+impl FileNameGenerator for DefaultFileNameGenerator {
+ fn generate_file_name(&self) -> String {
+ let file_id = self
+ .file_count
+ .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+ format!(
+ "{}-{:05}{}.{}",
+ self.prefix, file_id, self.suffix, self.format
+ )
+ }
+}
+
+#[cfg(test)]
Review Comment:
Can we add some tests here how what the filenames would look like?
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475447615
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
+ Self::MIN_BUFFER_SIZE
+ } else {
+ self.init_buffer_size
+ };
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ file_io: self.file_io,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+ file_io: FileIO,
+ field_ids: Vec<i32>,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ field_ids: &[i32],
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> Result<DataFileBuilder> {
+ // Only enter here when the file is not empty.
+ assert!(!metadata.row_groups.is_empty());
+ if field_ids.len() != metadata.row_groups[0].columns.len() {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ "len of field id is not match with len of columns in parquet metadata",
+ ));
+ }
+
+ let (column_sizes, value_counts, null_value_counts) =
+ {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group.columns.iter().zip(field_ids.iter()).for_each(
+ |(column_chunk, &field_id)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num.entry(field_id).or_insert(0_u64) +=
+ column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0) as u64;
+ }
+ },
+ )
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_path(file_path)
+ .file_format(DataFileFormat::Parquet)
+ .record_count(metadata.num_rows as u64)
+ .file_size_in_bytes(written_size as u64)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ // # TODO
+ // - nan_value_counts
+ // - lower_bounds
+ // - upper_bounds
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ Ok(builder)
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "write error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "close error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ if self.current_row_num == 0 {
Review Comment:
We should close the writer first. Otherwise, the writer's behavior is undefined.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475596750
##########
crates/iceberg/src/error.rs:
##########
@@ -44,6 +44,9 @@ pub enum ErrorKind {
///
/// This error is returned when given iceberg feature is not supported.
FeatureUnsupported,
+ /// Iceberg meets an unexpected parameter. Please refer the related doc or error message to
+ /// check if the parameter is valid.
+ UnexpectedParameter,
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475446762
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
> This comment is a little confusing, is the result a full path or related to table location?
A rel path(include data dir)for data file. e.g. data/xxx.parquet. I refine it later.
> Is this similar to LocationProvider in java API?
No. I find that the interface is not similar. `LocationProvider` seems a higher level abstract. It return a fully-qualified data file location for the given filename so that we also need to provide a file name.
The `LocationGenerator` will generate a fully-qualified file 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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1476988389
##########
crates/iceberg/src/writer/file_writer/mod.rs:
##########
@@ -18,16 +18,20 @@
//! This module contains the writer for data file format supported by iceberg: parquet, orc.
use super::{CurrentFileStatus, DefaultOutput};
-use crate::Result;
+use crate::{io::OutputFile, Result};
use arrow_array::RecordBatch;
use futures::Future;
+mod parquet_writer;
+pub use parquet_writer::{ParquetWriter, ParquetWriterBuilder};
+mod track_writer;
+
/// File writer builder trait.
pub trait FileWriterBuilder<O = DefaultOutput>: Send + Clone + 'static {
/// The associated file writer type.
type R: FileWriter<O>;
/// Build file writer.
- fn build(self) -> impl Future<Output = Result<Self::R>> + Send;
+ fn build(self, out_file: OutputFile) -> impl Future<Output = Result<Self::R>> + Send;
Review Comment:
I find a way to move it back.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1970336526
> I still question whether following Java's API is a good idea. I believe we can create a better and more user-friendly API for rust users.
In fact this is not end user api, but building blocks of end user api, e.g. we are supposed to build user friendly api on top of this. I agree that we should not blindly follow java's api, and provide more rusty api. But let's move first and improve it step by step.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492446872
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,385 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
Review Comment:
Yes, but seperating them can let user use `DefaultLocationGenerator` and `CustomFileNameGenertor` or other combinations.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470665655
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
> Does [writer in opendal](https://docs.rs/opendal/latest/opendal/struct.Operator.html?search=writer#method.writer) create the file?
It depends. AFAIK, opendal on s3 won't create file unless `close` has returned `Ok()`.
By the way, the use case you mentioned here more like [`Writer::abort`](https://opendal.apache.org/docs/rust/opendal/struct.Writer.html#method.abort).
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470668094
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
> By the way, the use case you mentioned here more like [Writer::abort](https://opendal.apache.org/docs/rust/opendal/struct.Writer.html#method.abort).
Nice. We replace delete with `abort`.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475452336
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
+ Self::MIN_BUFFER_SIZE
+ } else {
+ self.init_buffer_size
+ };
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ file_io: self.file_io,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
Review Comment:
I'm not sure what this means. ParquetWriter doesn't need another FileWriter. It's the lowest-level writer.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470721802
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
> I find that the `AsyncArrowWriter` already provides the [write with a buffer](https://docs.rs/parquet/latest/parquet/arrow/async_writer/struct.AsyncArrowWriter.html#method.write). In this case, maybe we can use this buffer.
Makes sense,
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1477561649
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
Review Comment:
We also need to add `partition_value` as parameter? This may affect the absolute path of file.
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
Review Comment:
```suggestion
/// Generate an absolute path for the given file name.
```
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
+}
+
+/// `FileNameGeneratorTrait` used to generate file name for data file. The file name can be passed to `LocationGenerator` to generate the location of the file.
+pub trait FileNameGenerator: Clone + Send + 'static {
+ /// Generate a file name.
+ fn generate_file_name(&self) -> String;
+}
+
+/// `DefaultFileNameGenerator` used to generate file name for data file. The file name can be
+/// passed to `LocationGenerator` to generate the location of the file.
+/// The rule of file name is aligned with the OutputFileFactory in iceberg-java
+#[derive(Clone)]
+pub struct DefaultFileNameGenerator {
+ partition_id: u64,
+ task_id: u64,
+ // The purpose of this id is to be able to know from two paths that they were written by the
+ // same operation.
+ // That's useful, for example, if a Spark job dies and leaves files in the file system, you can
+ // identify them all
+ // with a recursive listing and grep.
+ operator_id: String,
+ suffix: String,
Review Comment:
It's safer to use `DataFileFormat`?
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
+}
+
+/// `FileNameGeneratorTrait` used to generate file name for data file. The file name can be passed to `LocationGenerator` to generate the location of the file.
+pub trait FileNameGenerator: Clone + Send + 'static {
+ /// Generate a file name.
+ fn generate_file_name(&self) -> String;
+}
+
+/// `DefaultFileNameGenerator` used to generate file name for data file. The file name can be
+/// passed to `LocationGenerator` to generate the location of the file.
+/// The rule of file name is aligned with the OutputFileFactory in iceberg-java
+#[derive(Clone)]
+pub struct DefaultFileNameGenerator {
+ partition_id: u64,
Review Comment:
I'm hesitating to add these concepts `partition_id`, `task_id`, `operator_id` in the core crate, since these are all computation related concepts. As default, I this a random uuid would be enough. Compute engines can always override this trait.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1477719264
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
Review Comment:
Sounds reasonable to me.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492447476
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,385 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
Review Comment:
Yes, but seperating them can let user use `DefaultLocationGenerator` and `CustomFileNameGenertor` or other combinations.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1928766039
I think this writer is ready to go now. PTAL @Fokko
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475581238
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
Review Comment:
The reason is mainly about performance. Usually generics have better performance, but it may make it difficult to use for user. In this case I think it's reasonable to use generics, because it's not used by user directly.
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
+ Self::MIN_BUFFER_SIZE
+ } else {
+ self.init_buffer_size
+ };
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ file_io: self.file_io,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
Review Comment:
Same as below.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475583716
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
I feel the `LocationProvider` interface make more sense to me: it asks user to provide filename and partition value. We can't guess file suffix from table property, I think to refine the interface, we should ask user to provide things such as `DataFileFormat`, partition value.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470624504
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
I'm thinking of the bad cases: If the `delete` returns error, how can we retry this operation?
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470648930
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
Review Comment:
Sounds reasonable, we already have avro included, so it makes sense to include parquet.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1970299150
cc @Xuanwo Any other comments?
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1970342629
> My personal approach is to develop a simple, functional API that meets the required features and then build upon it. Having a powerful API from the start makes me nervous. Adding things is easy, but removing them is much harder.
In fact, such an api already has real world use case, for example, risingwave's streaming engine already uses such an api to write data.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475773886
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
For now, the writer doesn't need the LocationProvider. I will implement this `LocationProvider` in another 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: 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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470676713
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
> Yes, we can open a discussion about it. IMO, the concurrent writer is used in the case that writes multiple files concurrently. So concurrent writer may should be a higher-level writer.
Yes, there are two levels of concurrency here. We can write different files concurrently, and we can also write different parts of the same file concurrently. The second case is particularly relevant when writing a large parquet file, such as 10GiB or even 1TiB.
However, we don't need to delve into such details at the current stage. I just want to note that there are potential optimizations possible. Should we leave a todo 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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475592171
##########
crates/iceberg/src/error.rs:
##########
@@ -44,6 +44,9 @@ pub enum ErrorKind {
///
/// This error is returned when given iceberg feature is not supported.
FeatureUnsupported,
+ /// Iceberg meets an unexpected parameter. Please refer the related doc or error message to
+ /// check if the parameter is valid.
+ UnexpectedParameter,
Review Comment:
Oh, do you means that `ErrorKind` is used to identify the way to handle the error? For `Unexpect`, it means that iceberg can't handle this error and just return it. (I don't have a clear definition of what `ErrorKind` used to identify.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475621034
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
LGTM. How about keeping the same interface as `LocationProvider`? We can combine the `DataFileFormat` outside to keep the interface of `LocationProvider` simple.
--
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 parquet writer [iceberg-rust]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1508929875
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,594 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
+ )?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = max(Self::MIN_BUFFER_SIZE, self.init_buffer_size);
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to build parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+ field_ids: Vec<i32>,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ field_ids: &[i32],
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> Result<DataFileBuilder> {
+ // Only enter here when the file is not empty.
+ assert!(!metadata.row_groups.is_empty());
+ if field_ids.len() != metadata.row_groups[0].columns.len() {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ "Len of field id is not match with len of columns in parquet metadata.",
+ ));
+ }
+
+ let (column_sizes, value_counts, null_value_counts) =
+ {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group.columns.iter().zip(field_ids.iter()).for_each(
+ |(column_chunk, &field_id)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num.entry(field_id).or_insert(0_u64) +=
+ column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0) as u64;
+ }
+ },
+ )
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_path(file_path)
+ .file_format(DataFileFormat::Parquet)
+ .record_count(metadata.num_rows as u64)
+ .file_size_in_bytes(written_size as u64)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ // # TODO
+ // - nan_value_counts
+ // - lower_bounds
+ // - upper_bounds
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ Ok(builder)
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to write using parquet writer.",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to close parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
+
+ Ok(vec![Self::to_data_file_builder(
+ &self.field_ids,
+ metadata,
+ written_size as usize,
+ self.out_file.location().to_string(),
+ )?])
+ }
+}
+
+impl CurrentFileStatus for ParquetWriter {
+ fn current_file_path(&self) -> String {
+ self.out_file.location().to_string()
+ }
+
+ fn current_row_num(&self) -> usize {
+ self.current_row_num
+ }
+
+ fn current_written_size(&self) -> usize {
+ self.written_size.load(std::sync::atomic::Ordering::Relaxed) as usize
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use anyhow::Result;
+ use arrow_array::types::Int64Type;
+ use arrow_array::ArrayRef;
+ use arrow_array::Int64Array;
+ use arrow_array::RecordBatch;
+ use arrow_array::StructArray;
+ use bytes::Bytes;
+ use futures::AsyncReadExt;
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+ use tempfile::TempDir;
+
+ use super::*;
+ use crate::io::FileIOBuilder;
+ use crate::spec::Struct;
+ use crate::writer::file_writer::location_generator::test::MockLocationGenerator;
+ use crate::writer::file_writer::location_generator::DefaultFileNameGenerator;
+
+ #[derive(Clone)]
+ struct TestLocationGen;
+
+ #[tokio::test]
+ async fn test_parquet_writer() -> Result<()> {
+ let temp_dir = TempDir::new().unwrap();
+ let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+ let loccation_gen =
+ MockLocationGenerator::new(temp_dir.path().to_str().unwrap().to_string());
+ let file_name_gen =
+ DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet);
+
+ // prepare data
+ let schema = {
+ let fields = vec![
+ arrow_schema::Field::new("col", arrow_schema::DataType::Int64, true).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "0".to_string())]),
+ ),
+ ];
+ Arc::new(arrow_schema::Schema::new(fields))
+ };
+ let col = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
+ let null_col = Arc::new(Int64Array::new_null(1024)) as ArrayRef;
+ let to_write = RecordBatch::try_new(schema.clone(), vec![col]).unwrap();
+ let to_write_null = RecordBatch::try_new(schema.clone(), vec![null_col]).unwrap();
+
+ // write data
+ let mut pw = ParquetWriterBuilder::new(
+ 0,
+ WriterProperties::builder().build(),
+ to_write.schema(),
+ file_io.clone(),
+ loccation_gen,
+ file_name_gen,
+ )
+ .build()
+ .await?;
+ pw.write(&to_write).await?;
+ pw.write(&to_write_null).await?;
+ let res = pw.close().await?;
+ assert_eq!(res.len(), 1);
+ let data_file = res
+ .into_iter()
+ .next()
+ .unwrap()
+ // Put dummy field for build successfully.
+ .content(crate::spec::DataContentType::Data)
+ .partition(Struct::empty())
+ .build()
+ .unwrap();
+
+ // read the written file
+ let mut input_file = file_io
+ .new_input(data_file.file_path.clone())
+ .unwrap()
+ .reader()
+ .await
+ .unwrap();
+ let mut res = vec![];
+ let file_size = input_file.read_to_end(&mut res).await.unwrap();
+ let reader_builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(res)).unwrap();
+ let metadata = reader_builder.metadata().clone();
+
+ // check data
+ let mut reader = reader_builder.build().unwrap();
+ let res = reader.next().unwrap().unwrap();
+ assert_eq!(to_write, res);
+ let res = reader.next().unwrap().unwrap();
+ assert_eq!(to_write_null, res);
+
+ // check metadata
+ assert_eq!(metadata.num_row_groups(), 1);
+ assert_eq!(metadata.row_group(0).num_columns(), 1);
+ assert_eq!(data_file.file_format, DataFileFormat::Parquet);
+ assert_eq!(
+ data_file.record_count,
+ metadata
+ .row_groups()
+ .iter()
+ .map(|group| group.num_rows())
+ .sum::<i64>() as u64
+ );
+ assert_eq!(data_file.file_size_in_bytes, file_size as u64);
+ assert_eq!(data_file.column_sizes.len(), 1);
+ assert_eq!(
+ *data_file.column_sizes.get(&0).unwrap(),
+ metadata.row_group(0).column(0).compressed_size() as u64
+ );
+ assert_eq!(data_file.value_counts.len(), 1);
+ assert_eq!(*data_file.value_counts.get(&0).unwrap(), 2048);
+ assert_eq!(data_file.null_value_counts.len(), 1);
+ assert_eq!(*data_file.null_value_counts.get(&0).unwrap(), 1024);
+ assert_eq!(data_file.key_metadata.len(), 0);
+ assert_eq!(data_file.split_offsets.len(), 1);
+ assert_eq!(
+ *data_file.split_offsets.first().unwrap(),
+ metadata.row_group(0).file_offset().unwrap()
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_parquet_writer_with_complex_schema() -> Result<()> {
+ let temp_dir = TempDir::new().unwrap();
+ let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+ let loccation_gen =
Review Comment:
nit: typo: `location_gen`
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,594 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
+ )?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = max(Self::MIN_BUFFER_SIZE, self.init_buffer_size);
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to build parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+ field_ids: Vec<i32>,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ field_ids: &[i32],
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> Result<DataFileBuilder> {
+ // Only enter here when the file is not empty.
+ assert!(!metadata.row_groups.is_empty());
+ if field_ids.len() != metadata.row_groups[0].columns.len() {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ "Len of field id is not match with len of columns in parquet metadata.",
+ ));
+ }
+
+ let (column_sizes, value_counts, null_value_counts) =
+ {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group.columns.iter().zip(field_ids.iter()).for_each(
+ |(column_chunk, &field_id)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num.entry(field_id).or_insert(0_u64) +=
+ column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0) as u64;
+ }
+ },
+ )
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_path(file_path)
+ .file_format(DataFileFormat::Parquet)
+ .record_count(metadata.num_rows as u64)
+ .file_size_in_bytes(written_size as u64)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ // # TODO
+ // - nan_value_counts
+ // - lower_bounds
+ // - upper_bounds
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ Ok(builder)
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to write using parquet writer.",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to close parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
+
+ Ok(vec![Self::to_data_file_builder(
+ &self.field_ids,
+ metadata,
+ written_size as usize,
+ self.out_file.location().to_string(),
+ )?])
+ }
+}
+
+impl CurrentFileStatus for ParquetWriter {
+ fn current_file_path(&self) -> String {
+ self.out_file.location().to_string()
+ }
+
+ fn current_row_num(&self) -> usize {
+ self.current_row_num
+ }
+
+ fn current_written_size(&self) -> usize {
+ self.written_size.load(std::sync::atomic::Ordering::Relaxed) as usize
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use anyhow::Result;
+ use arrow_array::types::Int64Type;
+ use arrow_array::ArrayRef;
+ use arrow_array::Int64Array;
+ use arrow_array::RecordBatch;
+ use arrow_array::StructArray;
+ use bytes::Bytes;
+ use futures::AsyncReadExt;
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+ use tempfile::TempDir;
+
+ use super::*;
+ use crate::io::FileIOBuilder;
+ use crate::spec::Struct;
+ use crate::writer::file_writer::location_generator::test::MockLocationGenerator;
+ use crate::writer::file_writer::location_generator::DefaultFileNameGenerator;
+
+ #[derive(Clone)]
+ struct TestLocationGen;
+
+ #[tokio::test]
+ async fn test_parquet_writer() -> Result<()> {
+ let temp_dir = TempDir::new().unwrap();
+ let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+ let loccation_gen =
+ MockLocationGenerator::new(temp_dir.path().to_str().unwrap().to_string());
+ let file_name_gen =
+ DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet);
+
+ // prepare data
+ let schema = {
+ let fields = vec![
+ arrow_schema::Field::new("col", arrow_schema::DataType::Int64, true).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "0".to_string())]),
+ ),
+ ];
+ Arc::new(arrow_schema::Schema::new(fields))
+ };
+ let col = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
+ let null_col = Arc::new(Int64Array::new_null(1024)) as ArrayRef;
+ let to_write = RecordBatch::try_new(schema.clone(), vec![col]).unwrap();
+ let to_write_null = RecordBatch::try_new(schema.clone(), vec![null_col]).unwrap();
+
+ // write data
+ let mut pw = ParquetWriterBuilder::new(
+ 0,
+ WriterProperties::builder().build(),
+ to_write.schema(),
+ file_io.clone(),
+ loccation_gen,
+ file_name_gen,
+ )
+ .build()
+ .await?;
+ pw.write(&to_write).await?;
+ pw.write(&to_write_null).await?;
+ let res = pw.close().await?;
+ assert_eq!(res.len(), 1);
+ let data_file = res
+ .into_iter()
+ .next()
+ .unwrap()
+ // Put dummy field for build successfully.
+ .content(crate::spec::DataContentType::Data)
+ .partition(Struct::empty())
+ .build()
+ .unwrap();
+
+ // read the written file
+ let mut input_file = file_io
+ .new_input(data_file.file_path.clone())
+ .unwrap()
+ .reader()
+ .await
+ .unwrap();
+ let mut res = vec![];
+ let file_size = input_file.read_to_end(&mut res).await.unwrap();
+ let reader_builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(res)).unwrap();
+ let metadata = reader_builder.metadata().clone();
+
+ // check data
+ let mut reader = reader_builder.build().unwrap();
+ let res = reader.next().unwrap().unwrap();
+ assert_eq!(to_write, res);
+ let res = reader.next().unwrap().unwrap();
+ assert_eq!(to_write_null, res);
+
+ // check metadata
+ assert_eq!(metadata.num_row_groups(), 1);
+ assert_eq!(metadata.row_group(0).num_columns(), 1);
+ assert_eq!(data_file.file_format, DataFileFormat::Parquet);
+ assert_eq!(
+ data_file.record_count,
+ metadata
+ .row_groups()
+ .iter()
+ .map(|group| group.num_rows())
+ .sum::<i64>() as u64
+ );
+ assert_eq!(data_file.file_size_in_bytes, file_size as u64);
+ assert_eq!(data_file.column_sizes.len(), 1);
+ assert_eq!(
+ *data_file.column_sizes.get(&0).unwrap(),
+ metadata.row_group(0).column(0).compressed_size() as u64
+ );
+ assert_eq!(data_file.value_counts.len(), 1);
+ assert_eq!(*data_file.value_counts.get(&0).unwrap(), 2048);
+ assert_eq!(data_file.null_value_counts.len(), 1);
+ assert_eq!(*data_file.null_value_counts.get(&0).unwrap(), 1024);
+ assert_eq!(data_file.key_metadata.len(), 0);
+ assert_eq!(data_file.split_offsets.len(), 1);
+ assert_eq!(
+ *data_file.split_offsets.first().unwrap(),
+ metadata.row_group(0).file_offset().unwrap()
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_parquet_writer_with_complex_schema() -> Result<()> {
Review Comment:
Thanks for adding this test! Can you make sure that you also assign the field-ids for the sub-fields?
```
parq /var/folders/22/yb9h2zd55ql37h4_50xkmg7r0000gn/T/.tmpAMDXHW/test-00000.parquet --schema
# Schema
<pyarrow._parquet.ParquetSchema object at 0x124b8f940>
required group field_id=-1 arrow_schema {
optional int64 field_id=0 col0;
optional group field_id=1 col1 {
optional int64 field_id=-1 sub_col;
}
optional binary field_id=2 col2 (String);
optional group field_id=3 col3 (List) {
repeated group field_id=-1 list {
optional int64 field_id=-1 item;
}
}
optional group field_id=4 col4 {
optional group field_id=-1 sub_col {
optional int64 field_id=-1 sub_sub_col;
}
}
}
```
It would also be good to assert if they are there.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475451458
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
Review Comment:
I can try. But what's the benefit of making LocationGenerator generic? Seems the `LocationGenerator` will not be called frequently.🤔
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1476921216
##########
crates/iceberg/src/writer/file_writer/mod.rs:
##########
@@ -18,16 +18,20 @@
//! This module contains the writer for data file format supported by iceberg: parquet, orc.
use super::{CurrentFileStatus, DefaultOutput};
-use crate::Result;
+use crate::{io::OutputFile, Result};
use arrow_array::RecordBatch;
use futures::Future;
+mod parquet_writer;
+pub use parquet_writer::{ParquetWriter, ParquetWriterBuilder};
+mod track_writer;
+
/// File writer builder trait.
pub trait FileWriterBuilder<O = DefaultOutput>: Send + Clone + 'static {
/// The associated file writer type.
type R: FileWriter<O>;
/// Build file writer.
- fn build(self) -> impl Future<Output = Result<Self::R>> + Send;
+ fn build(self, out_file: OutputFile) -> impl Future<Output = Result<Self::R>> + Send;
Review Comment:
Why we need to change the interface? As we discussed, why not put `FileIO` in builder? I guess this will not work for `RollingFileWriterBuilder`?
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1469148875
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Review Comment:
This interface looks weird because returning a type here can't let us determine the type in `ParquetWriter`. So I modify it to a specific type here temporarily.
I think if we want to make it generic, maybe we need to make `OutputFile` a trait and let OutputFile bind an associate type like🤔
```
trait OutputFile {
type writer: FileWriter;
...
}
```
cc @liurenjie1024
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470677827
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
> Sounds reasonable for me. So this means that the writer in opendal will not provide the buffer cache, right?
opendal doesn't add buffer for users by default. Users will need to use `op.writer_with(path).buffer(8 * 1024 * 1024)` to enable 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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1970332953
> I still question whether following Java's API is a good idea. I believe we can create a better and more user-friendly API for rust users.
>
> However, this shouldn't hold us back. Let's move.
Thanks! Yes, it's not necessary to follow Java's API. For here, I just find that this design can give users more power to custom and reuse something so I follow it. It's ok for me to modify it if there is a simpler design.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470662411
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
> I'm thinking of the bad cases: If the delete returns error, how can we retry this operation?
This delete operation used in the cases:
The writer created and it didn't write anything, so we deleted it to avoid the empty file. If delete fails here, can we just skip this rare case?🤔
But I find that there maybe a wrong hypothesis. Does [writer in opendal](https://docs.rs/opendal/latest/opendal/struct.Operator.html?search=writer#method.writer) create 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
Re: [PR] feat: add parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475440018
##########
crates/iceberg/src/error.rs:
##########
@@ -44,6 +44,9 @@ pub enum ErrorKind {
///
/// This error is returned when given iceberg feature is not supported.
FeatureUnsupported,
+ /// Iceberg meets an unexpected parameter. Please refer the related doc or error message to
+ /// check if the parameter is valid.
+ UnexpectedParameter,
Review Comment:
This `Unexpected` means that Iceberg don't know what happened here, and no actions other than just returning it back. I think this error is used in where error happen in external system?🤔
The `UnexpectedParameter` means that the iceberg know the parameter error.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1477651653
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
+}
+
+/// `FileNameGeneratorTrait` used to generate file name for data file. The file name can be passed to `LocationGenerator` to generate the location of the file.
+pub trait FileNameGenerator: Clone + Send + 'static {
+ /// Generate a file name.
+ fn generate_file_name(&self) -> String;
+}
+
+/// `DefaultFileNameGenerator` used to generate file name for data file. The file name can be
+/// passed to `LocationGenerator` to generate the location of the file.
+/// The rule of file name is aligned with the OutputFileFactory in iceberg-java
+#[derive(Clone)]
+pub struct DefaultFileNameGenerator {
+ partition_id: u64,
Review Comment:
Sounds reasonable for me.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475578608
##########
crates/iceberg/src/error.rs:
##########
@@ -44,6 +44,9 @@ pub enum ErrorKind {
///
/// This error is returned when given iceberg feature is not supported.
FeatureUnsupported,
+ /// Iceberg meets an unexpected parameter. Please refer the related doc or error message to
+ /// check if the parameter is valid.
+ UnexpectedParameter,
Review Comment:
I still don't get the point. If users want to know the exact error, they can infer it from error message. But they still can't handle 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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1986843324
cc @ZENOTME Would you mind to fix the conflicts?
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1986854100
> > It would also be good to assert if they are there.
>
> Sorry, I'm not sure what this means. Did you mean assert like following?
>
> ```
> let schema = ...;
> assert_eq(schema.fields[0].metadata().get(field_id),1);
> ```
I fix it now. How about merging it first and refining the test later? After merging it, we can continue the writer's work.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470623846
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
Review Comment:
The files supported by iceberg are limited, which is not as extensive as catalogs. I think it's best to keep them in the same crate for easier use and management of dependencies.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470626858
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
It's better to add a `buffer` for this writer to avoid `EntiryTooSmall` error from s3.
---
There is another thing we can do is adding `concurrent` here so we can upload parts concurrently. But we need to make a design on how users control the concurrency. Maybe we can discuss them later.
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
+ // #TODO
+ // Do we need to check if file exists?
Review Comment:
It depends on whether we want to consider it an error if the file doesn't exist.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470626858
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
It's better to add a `buffer` (for example, 8MiB) for this writer to avoid `EntiryTooSmall` error from s3.
And in the future, we can add a config for this to allow users to control them.
---
There is another thing we can do is adding `concurrent` here so we can upload parts concurrently. But we need to make a design on how users control the concurrency. Maybe we can discuss them later.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470518241
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
Review Comment:
I'm thinking moving this to another crate `iceberg-parquet`, the core crate only contains `FileWriter` trait. WDYT? cc @ZENOTME @Xuanwo @Fokko
##########
crates/iceberg/src/spec/manifest.rs:
##########
@@ -924,7 +924,7 @@ impl TryFrom<i32> for ManifestStatus {
}
/// Data file carries data file path, partition tuple, metrics, …
-#[derive(Debug, PartialEq, Clone, Eq, TypedBuilder)]
+#[derive(Debug, PartialEq, Clone, Eq, Builder)]
Review Comment:
Why this change? `TypedBuilder` checks it in compile time, while `Builder` checks result in runtime.
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Review Comment:
We can't make it concrete `Writer` here since we can't expose opendal api to user. You can store `Box<dyn FileWriter>` in parquet writer.
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
This seems unsound to me. If it's deleted, we should consume it, so its argument should be `self` rather `&self`
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,267 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use opendal::Writer;
+use parquet::file::properties::WriterProperties;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+
+use super::{track_writer::TrackWriter, FileWriter, FileWriterBuilder};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ out_file: OutputFile,
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+}
+
+impl ParquetWriterBuilder {
+ /// Create a new `ParquetWriterBuilder`
+ pub fn new(
+ out_file: OutputFile,
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ ) -> Self {
+ Self {
+ out_file,
+ init_buffer_size,
+ props,
+ schema,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ let written_size = Arc::new(AtomicI64::new(0));
+ let inner_writer = TrackWriter::new(self.out_file.writer().await?, written_size.clone());
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema,
+ self.init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ Ok(ParquetWriter {
+ out_file: self.out_file,
+ writer,
+ written_size,
+ current_row_num: 0,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter<Writer>>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> DataFileBuilder {
+ let (column_sizes, value_counts, null_value_counts) = {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group
+ .columns
+ .iter()
+ .enumerate()
+ .for_each(|(column_id, column_chunk)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(column_id as i32).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(column_id as i32).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num
+ .entry(column_id as i32)
+ .or_insert(0_u64) += column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0)
+ as u64;
+ }
+ })
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_format(DataFileFormat::Parquet)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ .file_size_in_bytes(written_size as u64)
+ .record_count(metadata.num_rows as u64)
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .file_path(file_path)
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ builder
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "write error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "close error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ if self.current_row_num == 0 {
+ self.out_file.delete().await?;
+ return Ok(vec![]);
+ }
+ let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
+
+ Ok(vec![Self::to_data_file_builder(
+ metadata,
+ written_size as usize,
+ self.out_file.location().to_string(),
+ )])
+ }
+}
+
+impl CurrentFileStatus for ParquetWriter {
+ fn current_file_path(&self) -> String {
+ self.out_file.location().to_string()
+ }
+
+ fn current_row_num(&self) -> usize {
+ self.current_row_num
+ }
+
+ fn current_written_size(&self) -> usize {
+ self.written_size.load(std::sync::atomic::Ordering::Relaxed) as usize
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use anyhow::Result;
+ use arrow_array::ArrayRef;
+ use arrow_array::Int64Array;
+ use arrow_array::RecordBatch;
+ use bytes::Bytes;
+ use futures::AsyncReadExt;
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ use tempfile::TempDir;
+
+ use super::*;
+ use crate::io::FileIOBuilder;
+
+ #[derive(Clone)]
+ struct TestLocationGen;
+
+ #[tokio::test]
+ async fn test_parquet_writer() -> Result<()> {
+ // create output file
+ let tmp_dir = TempDir::new().unwrap();
+ let full_path = format!("{}/{}", tmp_dir.path().to_str().unwrap(), "test.parquet");
+ let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+ let output_file = file_io.new_output(&full_path).unwrap();
+
+ // prepare data
+ let col = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
+ let to_write = RecordBatch::try_from_iter([("col", col)]).unwrap();
+
+ // write data
+ let mut pw = ParquetWriterBuilder::new(
+ output_file.clone(),
+ 0,
+ WriterProperties::builder().build(),
+ to_write.schema(),
+ )
+ .build()
+ .await?;
+ pw.write(&to_write).await?;
+ pw.write(&to_write).await?;
+ pw.close().await?;
Review Comment:
We should also add some checks for data file fields.
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,267 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use opendal::Writer;
+use parquet::file::properties::WriterProperties;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+
+use super::{track_writer::TrackWriter, FileWriter, FileWriterBuilder};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ out_file: OutputFile,
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+}
+
+impl ParquetWriterBuilder {
+ /// Create a new `ParquetWriterBuilder`
+ pub fn new(
+ out_file: OutputFile,
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ ) -> Self {
+ Self {
+ out_file,
+ init_buffer_size,
+ props,
+ schema,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ let written_size = Arc::new(AtomicI64::new(0));
+ let inner_writer = TrackWriter::new(self.out_file.writer().await?, written_size.clone());
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema,
+ self.init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ Ok(ParquetWriter {
+ out_file: self.out_file,
+ writer,
+ written_size,
+ current_row_num: 0,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter<Writer>>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> DataFileBuilder {
+ let (column_sizes, value_counts, null_value_counts) = {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group
+ .columns
+ .iter()
+ .enumerate()
+ .for_each(|(column_id, column_chunk)| {
Review Comment:
This is incorrect, the `column_id` should be `field_id` of schema.
##########
crates/iceberg/src/io.rs:
##########
@@ -245,7 +245,7 @@ pub trait FileWrite: AsyncWrite {}
impl<T> FileWrite for T where T: AsyncWrite {}
/// Output file is used for writing to files..
-#[derive(Debug)]
+#[derive(Debug, Clone)]
Review Comment:
Is it possible to remove `Clone`? It's unsound to me to clone an `OutputFile`.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470626858
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
It's better to add a `buffer` (for example, 8MiB)for this writer to avoid `EntiryTooSmall` error from s3.
---
There is another thing we can do is adding `concurrent` here so we can upload parts concurrently. But we need to make a design on how users control the concurrency. Maybe we can discuss them later.
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
It's better to add a `buffer` (for example, 8MiB) for this writer to avoid `EntiryTooSmall` error from s3.
---
There is another thing we can do is adding `concurrent` here so we can upload parts concurrently. But we need to make a design on how users control the concurrency. Maybe we can discuss them later.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470712134
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
I find that the `AsyncArrowWriter` already provides the [write with a buffer](https://docs.rs/parquet/latest/parquet/arrow/async_writer/struct.AsyncArrowWriter.html#method.write ). In this case, maybe we can use this buffer.
And we also provide the interface `pub async fn writer(&self, buffer_size: usize) -> Result<Box<dyn FileWrite>>` in OutFile, for the file format which doesn't provide the cache, we can create a inner writer with buffer.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475635301
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
Sounds reasonable to me.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1473864622
##########
crates/iceberg/src/error.rs:
##########
@@ -44,6 +44,9 @@ pub enum ErrorKind {
///
/// This error is returned when given iceberg feature is not supported.
FeatureUnsupported,
+ /// Iceberg meets an unexpected parameter. Please refer the related doc or error message to
+ /// check if the parameter is valid.
+ UnexpectedParameter,
Review Comment:
I think this is similar to `Unexpected`?
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
Review Comment:
```suggestion
"Filed id not found in arrow schema metadata.",
```
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
Review Comment:
To improve readability, I would suggest to use alias `ArrowSchemaRef` here.
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
+ Self::MIN_BUFFER_SIZE
+ } else {
+ self.init_buffer_size
+ };
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
Review Comment:
```suggestion
"Failed to build parquet writer",
```
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
This comment is a little confusing, is the result a full path or related to table location?
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
+ Self::MIN_BUFFER_SIZE
+ } else {
+ self.init_buffer_size
+ };
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ file_io: self.file_io,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
Review Comment:
Is it possible to make `ParquetWriter<F: FileWriter>`? I think it's hidden under the `FileWriter` trait?
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -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.
+
+//! The module contains the location generator for file writer.
+
+use std::str::FromStr;
+use std::sync::atomic::AtomicUsize;
+use uuid::Uuid;
+
+use crate::spec::{DataFileFormat, TableMetadata};
+use crate::{Error, ErrorKind, Result};
+
+/// LocationGenerator will generate a file location for the writer.
+pub trait LocationGenerator: Send + Sync + 'static {
+ /// Generate a related file location for the writer.
Review Comment:
Is this similar to `LocationProvider` in java api?
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
Review Comment:
Is it possible to make `ParquetWriterBuilder` generics on `LocationGenerator`?
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
Review Comment:
```suggestion
let init_buffer_size = max(Self::MIN_BUFFER_SIZE, self.init_buffer_size);
```
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
Review Comment:
```suggestion
"Failed to parse field id",
```
##########
crates/iceberg/src/writer/file_writer/track_writer.rs:
##########
@@ -0,0 +1,72 @@
+// 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.
+
+use std::{
+ pin::Pin,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use tokio::io::AsyncWrite;
+
+use crate::io::FileWrite;
+
+/// `TrackWriter` is used to track the written size.
+pub struct TrackWriter {
Review Comment:
Do we need to expose this? I think `pub(crate)` would be enough?
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,390 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::Result;
+use crate::{
+ io::{FileIO, OutputFile},
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::LocationGenerator, track_writer::TrackWriter, FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+}
+
+impl ParquetWriterBuilder {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: SchemaRef,
+ file_io: FileIO,
+ location_generator: Arc<dyn LocationGenerator>,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ }
+ }
+}
+
+impl FileWriterBuilder for ParquetWriterBuilder {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id not found in schema",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::UnexpectedParameter,
+ "field id parse error",
+ )
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let file_path = self.location_generator.generate_name();
+ let out_file = self.file_io.new_output(file_path)?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = if self.init_buffer_size < Self::MIN_BUFFER_SIZE {
+ Self::MIN_BUFFER_SIZE
+ } else {
+ self.init_buffer_size
+ };
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "build error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ file_io: self.file_io,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+ file_io: FileIO,
+ field_ids: Vec<i32>,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ field_ids: &[i32],
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> Result<DataFileBuilder> {
+ // Only enter here when the file is not empty.
+ assert!(!metadata.row_groups.is_empty());
+ if field_ids.len() != metadata.row_groups[0].columns.len() {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ "len of field id is not match with len of columns in parquet metadata",
+ ));
+ }
+
+ let (column_sizes, value_counts, null_value_counts) =
+ {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group.columns.iter().zip(field_ids.iter()).for_each(
+ |(column_chunk, &field_id)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num.entry(field_id).or_insert(0_u64) +=
+ column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0) as u64;
+ }
+ },
+ )
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_path(file_path)
+ .file_format(DataFileFormat::Parquet)
+ .record_count(metadata.num_rows as u64)
+ .file_size_in_bytes(written_size as u64)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ // # TODO
+ // - nan_value_counts
+ // - lower_bounds
+ // - upper_bounds
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ Ok(builder)
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "write error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "close error from parquet writer",
+ )
+ .with_source(err)
+ })?;
+ if self.current_row_num == 0 {
Review Comment:
We should move this to first part?
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1986843681
> cc @ZENOTME Would you mind to fix the conflicts?
Sorry, don't notice it before. I fix it now.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1986860460
> > > It would also be good to assert if they are there.
>
> >
>
> > Sorry, I'm not sure what this means. Did you mean assert like following?
>
> >
>
> > ```
>
> > let schema = ...;
>
> > assert_eq(schema.fields[0].metadata().get(field_id),1);
>
> > ```
>
>
>
> I have fixed it now. How about merging it first and refining the test later? After merging it, we can continue the writer's work.
Sounds reasonable to me, thanks!
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 merged PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1970337631
> For here, I just find that this design can give users more power to custom and reuse something so I follow it. It's ok for me to modify it if there is a simpler design.
Thank you for understanding.
My personal approach is to develop a simple, functional API that meets the required features and then build upon it. Having a powerful API from the start makes me nervous. Adding things is easy, but removing them is much harder.
Both designs should work well. Iceberg-rust is still in its `0.x` versions, allowing us to test and validate our design with real-world use cases. Perhaps ultimately, users don't care about this aspect at all.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492447476
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,385 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
Review Comment:
Yes, but seperating them can let user use `DefaultLocationGenerator` and `CustomFileNameGenertor` or other combinations.
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475593785
##########
crates/iceberg/src/error.rs:
##########
@@ -44,6 +44,9 @@ pub enum ErrorKind {
///
/// This error is returned when given iceberg feature is not supported.
FeatureUnsupported,
+ /// Iceberg meets an unexpected parameter. Please refer the related doc or error message to
+ /// check if the parameter is valid.
+ UnexpectedParameter,
Review Comment:
Yes, I think this is the motivation we don't have much error kinds.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1475772911
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
I find that we don't need to delete empty files in this writer. We can do this thing in the upper writer and make this writer be simple. This writer only responded to get an output file and write it.
And I find that for FileWriterBuilder should accept the output file when built. We hope to clone the builder and rebuild a writer using a new output file, which means that the builder state should be bound with any output 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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1477655462
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,108 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
+ /// Generate a absolute path for the given file name.
+ /// e.g
+ /// For file name "part-00000.parquet", the generated location maybe "/table/data/part-00000.parquet"
+ fn generate_location(&self, file_name: &str) -> String;
Review Comment:
The reason why iceberg-java need this parameter is that the design of iceberg-java is to pass the `OutPutFile` to lower level writer. So that it need to generate the file path to lower level writer and the file path may have different partition dir.
But in our design, we let the lower level writer generate the `OutPutFile` using `FileIO` and `LocationGenerator` by itself. So I find that we can just use a LocationGenerator bind with a partition value, like
```
struct PartitionLocationGenerator {
dir: String
}
impl PartitionLocationGenerator {
pub fn new(partition_value: Value) -> Self
}
```
And do not need to use partition_value parameter.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1483994701
##########
crates/iceberg/src/io.rs:
##########
@@ -240,9 +241,9 @@ impl InputFile {
}
/// Trait for writing file.
-pub trait FileWrite: AsyncWrite {}
+pub trait FileWrite: AsyncWrite + TokioAsyncWrite + Send + Unpin {}
Review Comment:
It's quite weird that `FileWrite` requires to have both `AsyncWrite` and `TokioAsyncWrite`. It increases the burden for the downstream to implement.
Did I missed something here?
##########
crates/iceberg/src/writer/file_writer/location_generator.rs:
##########
@@ -0,0 +1,99 @@
+// 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.
+
+//! This module contains the location generator and file name generator for generating path of data file.
+
+use std::sync::{atomic::AtomicU64, Arc};
+
+use crate::spec::DataFileFormat;
+
+/// `LocationGenerator` used to generate the location of data file.
+pub trait LocationGenerator: Clone + Send + 'static {
Review Comment:
Can we merge `LocationGenerator` and `FileNameGenerator`? Both of them only have one easy function to implement. By merging them, we can reduce an extra generic param for users to use.
Or maybe we can go deeper by generating the final location in one shot.
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,385 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
Review Comment:
For example, we seem to always use them together.
##########
crates/iceberg/src/writer/file_writer/track_writer.rs:
##########
@@ -0,0 +1,72 @@
+// 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.
+
+use std::{
+ pin::Pin,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use tokio::io::AsyncWrite;
+
+use crate::io::FileWrite;
+
+/// `TrackWriter` is used to track the written size.
+pub(crate) struct TrackWriter {
+ inner: Box<dyn FileWrite>,
+ written_size: Arc<AtomicI64>,
Review Comment:
Writer only has `&mut` API so we can simplely use an `i64` here instead.
--
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 parquet writer [iceberg-rust]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492439815
##########
crates/iceberg/src/io.rs:
##########
@@ -240,9 +241,9 @@ impl InputFile {
}
/// Trait for writing file.
-pub trait FileWrite: AsyncWrite {}
+pub trait FileWrite: AsyncWrite + TokioAsyncWrite + Send + Unpin {}
Review Comment:
> Can we remove the futures::AsyncWrite? Since for now, we don't need it.
That's fine to remove `futures::AsyncWrite` here if not needed.
But I want to mention that using `AsyncWrite` is more friendly for users who don't want to depend on `TokioAsyncWrite` and there are `async_compat` crate to transform a `futures::AsyncWrite` to `TokioAsyncWrite`.
Maybe we can add our own trait to support both of them? I believe this is another issue and should not be part of this PR.
##########
crates/iceberg/src/io.rs:
##########
@@ -240,9 +241,9 @@ impl InputFile {
}
/// Trait for writing file.
-pub trait FileWrite: AsyncWrite {}
+pub trait FileWrite: AsyncWrite + TokioAsyncWrite + Send + Unpin {}
Review Comment:
> Can we remove the futures::AsyncWrite? Since for now, we don't need it.
That's fine to remove `futures::AsyncWrite` here if not needed.
But I want to mention that using `AsyncWrite` is more friendly for users who don't want to depend on `TokioAsyncWrite` and there are `async_compat` crate to transform a `futures::AsyncWrite` to `TokioAsyncWrite`.
---
Maybe we can add our own trait to support both of them? I believe this is another issue and should not be part of this 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: 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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492434129
##########
crates/iceberg/src/io.rs:
##########
@@ -240,9 +241,9 @@ impl InputFile {
}
/// Trait for writing file.
-pub trait FileWrite: AsyncWrite {}
+pub trait FileWrite: AsyncWrite + TokioAsyncWrite + Send + Unpin {}
Review Comment:
I added TokioAsyncWrite because the parquet writer accepct a tokio writer. Can we remove the futures::AsyncWrite? Since for now, we don't need 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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1492450927
##########
crates/iceberg/src/writer/file_writer/track_writer.rs:
##########
@@ -0,0 +1,72 @@
+// 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.
+
+use std::{
+ pin::Pin,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use tokio::io::AsyncWrite;
+
+use crate::io::FileWrite;
+
+/// `TrackWriter` is used to track the written size.
+pub(crate) struct TrackWriter {
+ inner: Box<dyn FileWrite>,
+ written_size: Arc<AtomicI64>,
Review Comment:
The track writer will be pass to parquet writer and then we can't get the info on it. Thit written_size is used to share the value so that we can read it outside, like:
```
let written_size = ..;
let track_writer = TrackWriter(written_size.clone());
let parquet_writer = ParquetWriter(track_writer);
// We can use written_size to read the written size.
```
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1509234137
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -0,0 +1,594 @@
+// 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.
+
+//! The module contains the file writer for parquet file format.
+
+use std::{
+ cmp::max,
+ collections::HashMap,
+ sync::{atomic::AtomicI64, Arc},
+};
+
+use crate::{io::FileIO, Result};
+use crate::{
+ io::OutputFile,
+ spec::{DataFileBuilder, DataFileFormat},
+ writer::CurrentFileStatus,
+ Error,
+};
+use arrow_schema::SchemaRef as ArrowSchemaRef;
+use parquet::{arrow::AsyncArrowWriter, format::FileMetaData};
+use parquet::{arrow::PARQUET_FIELD_ID_META_KEY, file::properties::WriterProperties};
+
+use super::{
+ location_generator::{FileNameGenerator, LocationGenerator},
+ track_writer::TrackWriter,
+ FileWriter, FileWriterBuilder,
+};
+
+/// ParquetWriterBuilder is used to builder a [`ParquetWriter`]
+#[derive(Clone)]
+pub struct ParquetWriterBuilder<T: LocationGenerator, F: FileNameGenerator> {
+ /// `buffer_size` determines the initial size of the intermediate buffer.
+ /// The intermediate buffer will automatically be resized if necessary
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> ParquetWriterBuilder<T, F> {
+ /// To avoid EntiryTooSmall error, we set the minimum buffer size to 8MB if the given buffer size is smaller than it.
+ const MIN_BUFFER_SIZE: usize = 8 * 1024 * 1024;
+
+ /// Create a new `ParquetWriterBuilder`
+ /// To construct the write result, the schema should contain the `PARQUET_FIELD_ID_META_KEY` metadata for each field.
+ pub fn new(
+ init_buffer_size: usize,
+ props: WriterProperties,
+ schema: ArrowSchemaRef,
+ file_io: FileIO,
+ location_generator: T,
+ file_name_generator: F,
+ ) -> Self {
+ Self {
+ init_buffer_size,
+ props,
+ schema,
+ file_io,
+ location_generator,
+ file_name_generator,
+ }
+ }
+}
+
+impl<T: LocationGenerator, F: FileNameGenerator> FileWriterBuilder for ParquetWriterBuilder<T, F> {
+ type R = ParquetWriter;
+
+ async fn build(self) -> crate::Result<Self::R> {
+ // Fetch field id from schema
+ let field_ids = self
+ .schema
+ .fields()
+ .iter()
+ .map(|field| {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .ok_or_else(|| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Field id not found in arrow schema metadata.",
+ )
+ })?
+ .parse::<i32>()
+ .map_err(|err| {
+ Error::new(crate::ErrorKind::Unexpected, "Failed to parse field id.")
+ .with_source(err)
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+
+ let written_size = Arc::new(AtomicI64::new(0));
+ let out_file = self.file_io.new_output(
+ self.location_generator
+ .generate_location(&self.file_name_generator.generate_file_name()),
+ )?;
+ let inner_writer = TrackWriter::new(out_file.writer().await?, written_size.clone());
+ let init_buffer_size = max(Self::MIN_BUFFER_SIZE, self.init_buffer_size);
+ let writer = AsyncArrowWriter::try_new(
+ inner_writer,
+ self.schema.clone(),
+ init_buffer_size,
+ Some(self.props),
+ )
+ .map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to build parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ Ok(ParquetWriter {
+ writer,
+ written_size,
+ current_row_num: 0,
+ out_file,
+ field_ids,
+ })
+ }
+}
+
+/// `ParquetWriter`` is used to write arrow data into parquet file on storage.
+pub struct ParquetWriter {
+ out_file: OutputFile,
+ writer: AsyncArrowWriter<TrackWriter>,
+ written_size: Arc<AtomicI64>,
+ current_row_num: usize,
+ field_ids: Vec<i32>,
+}
+
+impl ParquetWriter {
+ fn to_data_file_builder(
+ field_ids: &[i32],
+ metadata: FileMetaData,
+ written_size: usize,
+ file_path: String,
+ ) -> Result<DataFileBuilder> {
+ // Only enter here when the file is not empty.
+ assert!(!metadata.row_groups.is_empty());
+ if field_ids.len() != metadata.row_groups[0].columns.len() {
+ return Err(Error::new(
+ crate::ErrorKind::Unexpected,
+ "Len of field id is not match with len of columns in parquet metadata.",
+ ));
+ }
+
+ let (column_sizes, value_counts, null_value_counts) =
+ {
+ let mut per_col_size: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_val_num: HashMap<i32, u64> = HashMap::new();
+ let mut per_col_null_val_num: HashMap<i32, u64> = HashMap::new();
+ metadata.row_groups.iter().for_each(|group| {
+ group.columns.iter().zip(field_ids.iter()).for_each(
+ |(column_chunk, &field_id)| {
+ if let Some(column_chunk_metadata) = &column_chunk.meta_data {
+ *per_col_size.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.total_compressed_size as u64;
+ *per_col_val_num.entry(field_id).or_insert(0) +=
+ column_chunk_metadata.num_values as u64;
+ *per_col_null_val_num.entry(field_id).or_insert(0_u64) +=
+ column_chunk_metadata
+ .statistics
+ .as_ref()
+ .map(|s| s.null_count)
+ .unwrap_or(None)
+ .unwrap_or(0) as u64;
+ }
+ },
+ )
+ });
+ (per_col_size, per_col_val_num, per_col_null_val_num)
+ };
+
+ let mut builder = DataFileBuilder::default();
+ builder
+ .file_path(file_path)
+ .file_format(DataFileFormat::Parquet)
+ .record_count(metadata.num_rows as u64)
+ .file_size_in_bytes(written_size as u64)
+ .column_sizes(column_sizes)
+ .value_counts(value_counts)
+ .null_value_counts(null_value_counts)
+ // # TODO
+ // - nan_value_counts
+ // - lower_bounds
+ // - upper_bounds
+ .key_metadata(metadata.footer_signing_key_metadata.unwrap_or_default())
+ .split_offsets(
+ metadata
+ .row_groups
+ .iter()
+ .filter_map(|group| group.file_offset)
+ .collect(),
+ );
+ Ok(builder)
+ }
+}
+
+impl FileWriter for ParquetWriter {
+ async fn write(&mut self, batch: &arrow_array::RecordBatch) -> crate::Result<()> {
+ self.current_row_num += batch.num_rows();
+ self.writer.write(batch).await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to write using parquet writer.",
+ )
+ .with_source(err)
+ })?;
+ Ok(())
+ }
+
+ async fn close(self) -> crate::Result<Vec<crate::spec::DataFileBuilder>> {
+ let metadata = self.writer.close().await.map_err(|err| {
+ Error::new(
+ crate::ErrorKind::Unexpected,
+ "Failed to close parquet writer.",
+ )
+ .with_source(err)
+ })?;
+
+ let written_size = self.written_size.load(std::sync::atomic::Ordering::Relaxed);
+
+ Ok(vec![Self::to_data_file_builder(
+ &self.field_ids,
+ metadata,
+ written_size as usize,
+ self.out_file.location().to_string(),
+ )?])
+ }
+}
+
+impl CurrentFileStatus for ParquetWriter {
+ fn current_file_path(&self) -> String {
+ self.out_file.location().to_string()
+ }
+
+ fn current_row_num(&self) -> usize {
+ self.current_row_num
+ }
+
+ fn current_written_size(&self) -> usize {
+ self.written_size.load(std::sync::atomic::Ordering::Relaxed) as usize
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use anyhow::Result;
+ use arrow_array::types::Int64Type;
+ use arrow_array::ArrayRef;
+ use arrow_array::Int64Array;
+ use arrow_array::RecordBatch;
+ use arrow_array::StructArray;
+ use bytes::Bytes;
+ use futures::AsyncReadExt;
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+ use tempfile::TempDir;
+
+ use super::*;
+ use crate::io::FileIOBuilder;
+ use crate::spec::Struct;
+ use crate::writer::file_writer::location_generator::test::MockLocationGenerator;
+ use crate::writer::file_writer::location_generator::DefaultFileNameGenerator;
+
+ #[derive(Clone)]
+ struct TestLocationGen;
+
+ #[tokio::test]
+ async fn test_parquet_writer() -> Result<()> {
+ let temp_dir = TempDir::new().unwrap();
+ let file_io = FileIOBuilder::new_fs_io().build().unwrap();
+ let loccation_gen =
+ MockLocationGenerator::new(temp_dir.path().to_str().unwrap().to_string());
+ let file_name_gen =
+ DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet);
+
+ // prepare data
+ let schema = {
+ let fields = vec![
+ arrow_schema::Field::new("col", arrow_schema::DataType::Int64, true).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "0".to_string())]),
+ ),
+ ];
+ Arc::new(arrow_schema::Schema::new(fields))
+ };
+ let col = Arc::new(Int64Array::from_iter_values(vec![1; 1024])) as ArrayRef;
+ let null_col = Arc::new(Int64Array::new_null(1024)) as ArrayRef;
+ let to_write = RecordBatch::try_new(schema.clone(), vec![col]).unwrap();
+ let to_write_null = RecordBatch::try_new(schema.clone(), vec![null_col]).unwrap();
+
+ // write data
+ let mut pw = ParquetWriterBuilder::new(
+ 0,
+ WriterProperties::builder().build(),
+ to_write.schema(),
+ file_io.clone(),
+ loccation_gen,
+ file_name_gen,
+ )
+ .build()
+ .await?;
+ pw.write(&to_write).await?;
+ pw.write(&to_write_null).await?;
+ let res = pw.close().await?;
+ assert_eq!(res.len(), 1);
+ let data_file = res
+ .into_iter()
+ .next()
+ .unwrap()
+ // Put dummy field for build successfully.
+ .content(crate::spec::DataContentType::Data)
+ .partition(Struct::empty())
+ .build()
+ .unwrap();
+
+ // read the written file
+ let mut input_file = file_io
+ .new_input(data_file.file_path.clone())
+ .unwrap()
+ .reader()
+ .await
+ .unwrap();
+ let mut res = vec![];
+ let file_size = input_file.read_to_end(&mut res).await.unwrap();
+ let reader_builder = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(res)).unwrap();
+ let metadata = reader_builder.metadata().clone();
+
+ // check data
+ let mut reader = reader_builder.build().unwrap();
+ let res = reader.next().unwrap().unwrap();
+ assert_eq!(to_write, res);
+ let res = reader.next().unwrap().unwrap();
+ assert_eq!(to_write_null, res);
+
+ // check metadata
+ assert_eq!(metadata.num_row_groups(), 1);
+ assert_eq!(metadata.row_group(0).num_columns(), 1);
+ assert_eq!(data_file.file_format, DataFileFormat::Parquet);
+ assert_eq!(
+ data_file.record_count,
+ metadata
+ .row_groups()
+ .iter()
+ .map(|group| group.num_rows())
+ .sum::<i64>() as u64
+ );
+ assert_eq!(data_file.file_size_in_bytes, file_size as u64);
+ assert_eq!(data_file.column_sizes.len(), 1);
+ assert_eq!(
+ *data_file.column_sizes.get(&0).unwrap(),
+ metadata.row_group(0).column(0).compressed_size() as u64
+ );
+ assert_eq!(data_file.value_counts.len(), 1);
+ assert_eq!(*data_file.value_counts.get(&0).unwrap(), 2048);
+ assert_eq!(data_file.null_value_counts.len(), 1);
+ assert_eq!(*data_file.null_value_counts.get(&0).unwrap(), 1024);
+ assert_eq!(data_file.key_metadata.len(), 0);
+ assert_eq!(data_file.split_offsets.len(), 1);
+ assert_eq!(
+ *data_file.split_offsets.first().unwrap(),
+ metadata.row_group(0).file_offset().unwrap()
+ );
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_parquet_writer_with_complex_schema() -> Result<()> {
Review Comment:
> It would also be good to assert if they are there.
Sorry, I'm not sure what this means. Did you mean assert like following?
```
let schema = ...;
assert_eq(schema.fields[0].metadata().get(field_id),1);
```
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#issuecomment-1972533382
I have added more tests. Feel free to tell me if there is still something need to improve. @Fokko
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470664532
##########
crates/iceberg/src/spec/manifest.rs:
##########
@@ -924,7 +924,7 @@ impl TryFrom<i32> for ManifestStatus {
}
/// Data file carries data file path, partition tuple, metrics, …
-#[derive(Debug, PartialEq, Clone, Eq, TypedBuilder)]
+#[derive(Debug, PartialEq, Clone, Eq, Builder)]
Review Comment:
Because I find that we need to return the Builder type, if we use `TypedBuilder`, we can't use `fn close(&self) -> Vec<Builder>` in trait, because the real return type maybe
`
fn close(&self) -> Vec<Builder<...>>
`
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470664109
##########
crates/iceberg/src/io.rs:
##########
@@ -268,6 +268,16 @@ impl OutputFile {
.await?)
}
+ /// Delete file.
+ pub async fn delete(&self) -> Result<()> {
Review Comment:
It seems more reasonable to move it to `FileIO`. An `OutputFile` may not exists yet.
--
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 parquet writer [iceberg-rust]
Posted by "ZENOTME (via GitHub)" <gi...@apache.org>.
ZENOTME commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470675094
##########
crates/iceberg/src/io.rs:
##########
@@ -278,7 +288,7 @@ impl OutputFile {
}
/// Creates output file for writing.
- pub async fn writer(&self) -> Result<impl FileWrite> {
+ pub async fn writer(&self) -> Result<Writer> {
Ok(self.op.writer(&self.path[self.relative_path_pos..]).await?)
Review Comment:
> It's better to add a buffer (for example, 8MiB) for this writer to avoid EntiryTooSmall error from s3.
> And in the future, we can add a config for this to allow users to control them.
Sounds reasonable for me. So this means that the writer in opendal will not provide the buffer cache, right?
--
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 parquet writer [iceberg-rust]
Posted by "liurenjie1024 (via GitHub)" <gi...@apache.org>.
liurenjie1024 commented on code in PR #176:
URL: https://github.com/apache/iceberg-rust/pull/176#discussion_r1470792480
##########
crates/iceberg/src/spec/manifest.rs:
##########
@@ -924,7 +924,7 @@ impl TryFrom<i32> for ManifestStatus {
}
/// Data file carries data file path, partition tuple, metrics, …
-#[derive(Debug, PartialEq, Clone, Eq, TypedBuilder)]
+#[derive(Debug, PartialEq, Clone, Eq, Builder)]
Review Comment:
Oh, I see, thanks!
--
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