diff --git a/python/pyarrow/_dataset_parquet.pyx b/python/pyarrow/_dataset_parquet.pyx index 6ac9383a022..fac8e96df85 100644 --- a/python/pyarrow/_dataset_parquet.pyx +++ b/python/pyarrow/_dataset_parquet.pyx @@ -660,6 +660,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions): write_page_checksum=self._properties["write_page_checksum"], sorting_columns=self._properties["sorting_columns"], store_decimal_as_integer=self._properties["store_decimal_as_integer"], + use_content_defined_chunking=self._properties["use_content_defined_chunking"], ) def _set_arrow_properties(self): @@ -713,6 +714,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions): write_page_checksum=False, sorting_columns=None, store_decimal_as_integer=False, + use_content_defined_chunking=False, ) self._set_properties() diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py index 8e49a96a8f8..152c2694e2a 100644 --- a/python/pyarrow/parquet/core.py +++ b/python/pyarrow/parquet/core.py @@ -806,10 +806,6 @@ def _sanitize_table(table, new_schema, flavor): Set a target threshold for the approximate encoded size of data pages within a column chunk (in bytes). If None, use the default data page size of 1MByte. -max_rows_per_page : int, default None - Maximum number of rows per page within a column chunk. - If None, use the default of 20000. - Smaller values reduce memory usage during reads but increase metadata overhead. flavor : {'spark'}, default None Sanitize schema or set other compatibility options to work with various target systems. @@ -924,6 +920,42 @@ def _sanitize_table(table, new_schema, flavor): - fixed_len_byte_array: for precision > 18. As a consequence, decimal columns stored in integer types are more compact. +write_time_adjusted_to_utc : bool, default False + Set the value of isAdjustedTOUTC when writing a TIME column. + If True, this tells the Parquet reader that the TIME columns + are expressed in reference to midnight in the UTC timezone. + If False (the default), the TIME columns are assumed to be expressed + in reference to midnight in an unknown, presumably local, timezone. +max_rows_per_page : int, default None + Maximum number of rows per page within a column chunk. + If None, use the default of 20000. + Smaller values reduce memory usage during reads but increase metadata overhead. +bloom_filter_options : dict, default None + Create Bloom filters for the columns specified by the provided `dict`. + + Bloom filters can be configured with two parameters: number of distinct values + (NDV), and false-positive probability (FPP). + + Bloom filters are most effective for high-cardinality columns. A good default + is to set NDV equal to the number of rows. Lower values reduce disk usage but + may not be worthwhile for very small NDVs. Increasing NDV (without increasing FPP) + increases disk and memory usage. + + Lower FPP values require more disk and memory space. For a fixed NDV, the + space requirement grows roughly proportional to log(1/FPP). Recommended + values are 0.1, 0.05, or 0.01. Very small values are counterproductive as + the bitset may exceed the size of the actual data. Set NDV appropriately + to minimize space usage. + + The keys of the `dict` are column paths. For each path, the value can be either: + + - A dictionary, with keys `ndv` and `fpp`. The value for `ndv` must be a positive + integer. If the 'ndv' key is not present, the default value of `1048576` will be + used. The value for `fpp` must be a float between 0.0 and 1.0. If the `fpp` key + is not present, the default value of `0.05` will be used. + - A boolean, with ``True`` indicating that a Bloom filter should be produced with + the above mentioned default values of `ndv=1048576` and `fpp=0.05`. This is + equivalent to passing an empty dict. use_content_defined_chunking : bool or dict, default False Optimize parquet files for content addressable storage (CAS) systems by writing data pages according to content-defined chunk boundaries. This allows for more @@ -958,39 +990,6 @@ def _sanitize_table(table, new_schema, flavor): balance between deduplication ratio and fragmentation. Use norm_level=1 or norm_level=2 to reach a higher deduplication ratio at the expense of fragmentation. - -write_time_adjusted_to_utc : bool, default False - Set the value of isAdjustedTOUTC when writing a TIME column. - If True, this tells the Parquet reader that the TIME columns - are expressed in reference to midnight in the UTC timezone. - If False (the default), the TIME columns are assumed to be expressed - in reference to midnight in an unknown, presumably local, timezone. -bloom_filter_options : dict, default None - Create Bloom filters for the columns specified by the provided `dict`. - - Bloom filters can be configured with two parameters: number of distinct values - (NDV), and false-positive probability (FPP). - - Bloom filters are most effective for high-cardinality columns. A good default - is to set NDV equal to the number of rows. Lower values reduce disk usage but - may not be worthwhile for very small NDVs. Increasing NDV (without increasing FPP) - increases disk and memory usage. - - Lower FPP values require more disk and memory space. For a fixed NDV, the - space requirement grows roughly proportional to log(1/FPP). Recommended - values are 0.1, 0.05, or 0.01. Very small values are counterproductive as - the bitset may exceed the size of the actual data. Set NDV appropriately - to minimize space usage. - - The keys of the `dict` are column paths. For each path, the value can be either: - - - A dictionary, with keys `ndv` and `fpp`. The value for `ndv` must be a positive - integer. If the 'ndv' key is not present, the default value of `1048576` will be - used. The value for `fpp` must be a float between 0.0 and 1.0. If the `fpp` key - is not present, the default value of `0.05` will be used. - - A boolean, with ``True`` indicating that a Bloom filter should be produced with - the above mentioned default values of `ndv=1048576` and `fpp=0.05`. This is - equivalent to passing an empty dict. """ _parquet_writer_example_doc = """\ @@ -1089,6 +1088,7 @@ def __init__(self, where, schema, filesystem=None, store_decimal_as_integer=False, write_time_adjusted_to_utc=False, max_rows_per_page=None, + use_content_defined_chunking=False, **options): if use_deprecated_int96_timestamps is None: # Use int96 timestamps for Spark @@ -1144,6 +1144,7 @@ def __init__(self, where, schema, filesystem=None, store_decimal_as_integer=store_decimal_as_integer, write_time_adjusted_to_utc=write_time_adjusted_to_utc, max_rows_per_page=max_rows_per_page, + use_content_defined_chunking=use_content_defined_chunking, **options) self.is_open = True @@ -2034,6 +2035,7 @@ def write_table(table, where, row_group_size=None, version='2.6', write_time_adjusted_to_utc=False, max_rows_per_page=None, bloom_filter_options=None, + use_content_defined_chunking=False, **kwargs): # Implementor's note: when adding keywords here / updating defaults, also # update it in write_to_dataset and _dataset_parquet.pyx ParquetFileWriteOptions @@ -2068,6 +2070,7 @@ def write_table(table, where, row_group_size=None, version='2.6', write_time_adjusted_to_utc=write_time_adjusted_to_utc, max_rows_per_page=max_rows_per_page, bloom_filter_options=bloom_filter_options, + use_content_defined_chunking=use_content_defined_chunking, **kwargs) as writer: writer.write_table(table, row_group_size=row_group_size) except Exception: diff --git a/python/pyarrow/tests/test_dataset.py b/python/pyarrow/tests/test_dataset.py index 1e0fa7c2e24..0a94c0bd987 100644 --- a/python/pyarrow/tests/test_dataset.py +++ b/python/pyarrow/tests/test_dataset.py @@ -5936,6 +5936,19 @@ def test_checksum_write_dataset_read_dataset_to_table(tempdir): ).to_table() +@pytest.mark.parquet +@pytest.mark.parametrize("cdc", [ + True, + {"min_chunk_size": 32 * 1024, "max_chunk_size": 64 * 1024}, +]) +def test_write_dataset_content_defined_chunking(tempdir, cdc): + expected_table = pa.table({'a': [1, 2, 3]}) + fmt = ds.ParquetFileFormat() + opts = fmt.make_write_options(use_content_defined_chunking=cdc) + ds.write_dataset(expected_table, tempdir, format=fmt, file_options=opts) + assert ds.dataset(tempdir, format=fmt).to_table().equals(expected_table) + + def test_make_write_options_error(): # GH-39440: calling make_write_options as a static class method msg_1 = ("make_write_options() should be called on an "