Repository: dask/dask
File: dask/dataframe/io/parquet/arrow.py (Python)

Code before:
            ]
            if partitions and dataset.metadata:
                # Dont use dataset.metadata for partitioned datasets, unless
                # the column-chunk metadata includes the `"file_path"`.
                # The order of dataset.metadata.row_group items is often
                # different than the order of `dataset.pieces`.
                if not col_chunk_paths or (
                    len(dataset.pieces) != dataset.metadata.num_row_groups
                ):
                    dataset.schema = dataset.metadata.schema
                    dataset.metadata = None
        else:
            partitions = []

        if dataset.metadata:
            schema = dataset.metadata.schema.to_arrow_schema()
        else:
            schema = dataset.schema.to_arrow_schema()
        columns = None

        has_pandas_metadata = (
            schema.metadata is not None and b"pandas" in schema.metadata
        )

        if has_pandas_metadata:
            pandas_metadata = json.loads(schema.metadata[b"pandas"].decode("utf8"))
            (
                index_names,
                column_names,
                storage_name_mapping,
                column_index_names,
            ) = _parse_pandas_metadata(pandas_metadata)
            if categories is None:
                categories = []
                for col in pandas_metadata["columns"]:
                    if (col["pandas_type"] == "categorical") and (
                        col["name"] not in categories
                    ):
                        categories.append(col["name"])
        else:
            index_names = []
            column_names = schema.names
            storage_name_mapping = {k: k for k in column_names}
            column_index_names = [None]

        if index is None and index_names:
            index = index_names

        if set(column_names).intersection(partitions):
            raise ValueError(
                "partition(s) should not exist in columns.\n"

Review comment (suggestion):
Might be worth to add a short comment here why this is done. My understanding is that this checks if the file paths are correctly set in the `_metadata` file (so which was not the case with dask+pyarrow right now)

Code after:

    return md_list


class ArrowEngine(Engine):
    @staticmethod
    def read_metadata(
        fs,
        paths,
        categories=None,
        index=None,
        gather_statistics=None,
        filters=None,
        split_row_groups=True,
        **kwargs,
    ):
        # Define the dataset object to use for metadata,
        # Also, initialize `parts`.  If `parts` is populated here,
        # then each part will correspond to a file.  Otherwise, each part will
        # correspond to a row group (populated below)
        parts, dataset = _determine_dataset_parts(
            fs, paths, gather_statistics, filters, kwargs.get("dataset", {})
        )
        # Check if the column-chunk file_path's are set in "_metadata".
        # If available, we can use the path to sort the row-groups
        col_chunk_paths = False
        if dataset.metadata:
            col_chunk_paths = all(
                dataset.metadata.row_group(i).column(0).file_path is not None
                for i in range(dataset.metadata.num_row_groups)
            )

        # TODO: Call to `_determine_dataset_parts` uses `pq.ParquetDataset`
        # to define the `dataset` object. `split_row_groups` should be passed
        # to that constructor once it is supported (see ARROW-2801).
        if dataset.partitions is not None:
            partitions = [
                n for n in dataset.partitions.partition_names if n is not None
            ]
            if partitions and dataset.metadata:
                # Dont use dataset.metadata for partitioned datasets, unless
                # the column-chunk metadata includes the `"file_path"`.
                # The order of dataset.metadata.row_group items is often
                # different than the order of `dataset.pieces`.
                if not col_chunk_paths or (
                    len(dataset.pieces) != dataset.metadata.num_row_groups
                ):
                    dataset.schema = dataset.metadata.schema
                    dataset.metadata = None
        else:
            partitions = []