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

Code before:

        # Leaving index as a column in `meta`, because the index
        # will be reset below (in case the index was detected after
        # meta was created)
        if index_in_columns:
            meta = meta[columns + index]
        else:
            meta = meta[columns]

    else:
        meta = meta[list(columns)]

    return meta, index, columns


def aggregate_row_groups(parts, stats, chunksize):
    if not (stats[0]["file_path_0"] and stats[0]["total_byte_size"]):
        return parts, stats

    chunksize = parse_bytes(chunksize)

    def _add_piece(part, part_list, stat, stat_list):
        part_list.append(part)
        stat_list.append(stat)

    def _update_piece(part, next_part, stat, next_stat):
        # Update part list
        next_part.append(part)

        # Update Statistics
        next_stat["total_byte_size"] += stat["total_byte_size"]
        next_stat["num-rows"] += stat["num-rows"]
        for col, col_add in zip(next_stat["columns"], stat["columns"]):
            if col["name"] != col_add["name"]:
                raise ValueError("Columns are different!!")
            if "null_count" in col:
                col["null_count"] += col_add["null_count"]
            if "min" in col:
                col["min"] = min(col["min"], col_add["min"])
            if "max" in col:
                col["max"] = max(col["max"], col_add["max"])

    parts_agg = []
    stats_agg = []
    next_part, next_stat = [parts[0].copy()], stats[0].copy()
    for i in range(1, len(parts)):
        stat, part = stats[i], parts[i]
        if (stat["file_path_0"] == next_stat["file_path_0"]) and (
            (next_stat["total_byte_size"] + stat["total_byte_size"]) <= chunksize
        ):
            _update_piece(part, next_part, stat, next_stat)

Review comment (suggestion):
Same here.  If you don't mind I would prefer that we didn't nest functions like this.  Rambling thoughts here: http://matthewrocklin.com/blog/work/2019/06/23/avoid-indirection

Code after:

    else:
        meta = meta[list(columns)]

    return meta, index, columns


def aggregate_row_groups(parts, stats, chunksize):
    if not stats[0]["file_path_0"]:
        return parts, stats

    parts_agg = []
    stats_agg = []
    chunksize = parse_bytes(chunksize)
    next_part, next_stat = [parts[0].copy()], stats[0].copy()
    for i in range(1, len(parts)):
        stat, part = stats[i], parts[i]
        if (stat["file_path_0"] == next_stat["file_path_0"]) and (
            (next_stat["total_byte_size"] + stat["total_byte_size"]) <= chunksize
        ):
            # Update part list
            next_part.append(part)

            # Update Statistics
            next_stat["total_byte_size"] += stat["total_byte_size"]
            next_stat["num-rows"] += stat["num-rows"]
            for col, col_add in zip(next_stat["columns"], stat["columns"]):
                if col["name"] != col_add["name"]:
                    raise ValueError("Columns are different!!")
                if "null_count" in col:
                    col["null_count"] += col_add["null_count"]
                if "min" in col:
                    col["min"] = min(col["min"], col_add["min"])
                if "max" in col:
                    col["max"] = max(col["max"], col_add["max"])
        else:
            parts_agg.append(next_part)
            stats_agg.append(next_stat)
            next_part, next_stat = [part.copy()], stat.copy()

    parts_agg.append(next_part)
    stats_agg.append(next_stat)

    return parts_agg, stats_agg


DataFrame.to_parquet.__doc__ = to_parquet.__doc__