Repository: dask/fastparquet

Issue:
OverflowError with a 3GB, 11M-line JSONL file
The following was run on Ubuntu 20 on a `e2-highcpu-32` GCP VM with 32 GB of RAM and 32 vCPUs.

I downloaded the California dataset from https://github.com/microsoft/USBuildingFootprints and converted it from JSONL into Parquet with pyarrow and I attempted to do the same with fastparquet.

```bash
$ ogr2ogr -f GeoJSONSeq /vsistdout/ California.geojson \
    | jq -c '.properties * {geom: .geometry|tostring}' \
    > California.jsonl
$ head -n1 California.jsonl | jq .
```

```json
{
  "release": 1,
  "capture_dates_range": "",
  "geom": "{\"type\":\"Polygon\",\"coordinates\":[[[-114.127454,34.265674],[-114.127476,34.265839],[-114.127588,34.265829],[-114.127565,34.265663],[-114.127454,34.265674]]]}"
}
```

PyArrow is able to produce a 792 MB Parquet file without issue.

```bash
$ python3 -c "import pandas as pd; pd.read_json('California.jsonl', lines=True).to_parquet('pandas.pyarrow.snappy.pq', engine='pyarrow')"
```

The resulting Parquet file is 792 MB with a single row group.

```
<pyarrow._parquet.FileMetaData object at 0x7f18566254a0>
  created_by: parquet-cpp-arrow version 10.0.1
  num_columns: 3
  num_rows: 11542912
  num_row_groups: 1
  format_version: 2.6
  serialized_size: 3064
```

When I attempt the same with fastparquet I'm hit with an `OverflowError`.

```bash
$ python3 -c "import pandas as pd; pd.read_json('California.jsonl', lines=True).to_parquet('pandas.fastparquet.snappy.pq', engine='fastparquet')"
```

These are the versions of software involved:

* pandas-1.5.2-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl
* fastparquet-2022.12.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl
* pyarrow-10.0.1-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl

Are there any flags I could pass to fastparquet to get this workload to run on a system with 32 GB of RAM?

Discussion:

Patch:
diff --git a/fastparquet/writer.py b/fastparquet/writer.py
index dbff11f..a26e49a 100644
--- a/fastparquet/writer.py
+++ b/fastparquet/writer.py
@@ -322,26 +322,39 @@ def convert(data, se):
 
 
 def infer_object_encoding(data):
-    head = data[:10] if isinstance(data, pd.Index) else data.dropna().iloc[:10]
-    if all(isinstance(i, str) for i in head if i is not None):
+    """Guess object type from first 10 non-na values by iteration"""
+    if data.empty:
         return "utf8"
-    elif all(isinstance(i, bytes) for i in head if i is not None):
-        return 'bytes'
-    elif all(isinstance(i, (list, dict)) for i in head if i is not None):
-        return 'json'
-    elif all(isinstance(i, bool) for i in head if i is not None):
-        return 'bool'
-    elif all(isinstance(i, Decimal) for i in head if i is not None):
-        return 'decimal'
-    elif all(isinstance(i, int) for i in head if i is not None):
-        return 'int'
-    elif all(isinstance(i, float) or isinstance(i, np.floating)
-             for i in head if i):
-        # You need np.floating here for pandas NaNs in object
-        # columns with python floats.
-        return 'float'
-    else:
-        raise ValueError("Can't infer object conversion type: %s" % head)
+    t = None
+    s = 0
+    encs = {
+        str: "utf8",
+        bytes: "bytes",
+        list: "json",
+        dict: "json",
+        bool: "bool",
+        Decimal: "decimal",
+        int: "int",
+        float: "float",
+        np.floating: "float",
+        np.str_: "utf8"
+    }
+    for i in data:
+        if pd.isna(i):
+            continue
+        tt = type(i)
+        if tt in encs:
+            tt = encs[tt]
+            if t is None:
+                t = tt
+            elif t != tt:
+                raise ValueError("Can't infer object conversion type: %s" % data)
+            s += 1
+        else:
+            raise ValueError("Can't infer object conversion type: %s" % data)
+        if s > 10:
+            break
+    return t
 
 
 def time_shift(indata, outdata, factor=1000):
@@ -400,7 +413,8 @@ def make_definitions(data, no_nulls, datapage_version=1):
         out = data
     else:
         se = parquet_thrift.SchemaElement(type=parquet_thrift.Type.BOOLEAN)
-        out = encode_plain(data.notnull(), se)
+        dnn = data.notnull()
+        out = encode_plain(dnn, se)
 
         cencoding.encode_unsigned_varint(len(out) << 1 | 1, temp)
         head = temp.so_far()
@@ -412,14 +426,39 @@ def make_definitions(data, no_nulls, datapage_version=1):
             # no need to write length, it's in the header
             # head.write(out)?
             block = bytes(head) + out
-        out = data.dropna()  # better, data[data.notnull()], from above ?
+        out = data[dnn]
     return block, out
 
 
 DATAPAGE_VERSION = 2 if os.environ.get("FASTPARQUET_DATAPAGE_V2", False) else 1
+MAX_PAGE_SIZE = 500 * 2**20
+
+
+def _rows_per_page(data, selement, has_nulls=True, page_size=None):
+    page_size = page_size or MAX_PAGE_SIZE
+    if is_categorical_dtype(data.dtype):
+        bytes_per_element = data.cat.codes.dtype.itemsize
+    elif selement.type == parquet_thrift.Type.BOOLEAN:
+        bytes_per_element = 0.125
+    elif selement.type == parquet_thrift.Type.INT64:
+        bytes_per_element = 8
+    elif selement.type == parquet_thrift.Type.INT32:
+        bytes_per_element = 4
+    elif isinstance(data.dtype, BaseMaskedDtype) and data.dtype in pdoptional_to_numpy_typemap:
+        bytes_per_element = np.dtype(pdoptional_to_numpy_typemap[data.dtype]).itemsize
+    elif data.dtype == "object" or str(data.dtype) == "string":
+        dd = data.iloc[:1000]
+        d2 = dd[dd.notnull()]
+        sample = d2.astype(str).map(len)
+        chrs = sample.sum()
+        bytes_per_element = chrs / (len(sample) or 4) + 4
+    else:
+        bytes_per_element = data.dtype.itemsize
+
+    return int(page_size // (bytes_per_element + 0.125 * has_nulls))
 
 
-def write_column(f, data, selement, compression=None, datapage_version=None,
+def write_column(f, data0, selement, compression=None, datapage_version=None,
                  stats=True):
     """
     Write a single column of data to an open Parquet file
@@ -427,7 +466,7 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
     Parameters
     ----------
     f: open binary file
-    data: pandas Series or numpy (1d) array
+    data0: pandas Series
     selement: thrift SchemaElement
         produced by ``find_type``
     compression: str, dict, or None
@@ -450,177 +489,186 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
     """
     datapage_version = datapage_version or DATAPAGE_VERSION
     has_nulls = selement.repetition_type == parquet_thrift.FieldRepetitionType.OPTIONAL
-    tot_rows = len(data)
+    tot_rows = len(data0)
     encoding = "PLAIN"
-
-    if has_nulls:
-        if is_categorical_dtype(data.dtype):
-            num_nulls = (data.cat.codes == -1).sum()
-        else:
-            num_nulls = len(data) - data.count()
-        definition_data, data = make_definitions(data, num_nulls == 0, datapage_version=datapage_version)
-        # make_definitions returns `data` with all nulls dropped
-        # the null-stripped `data` can be converted from Optional Types to
-        # their numpy counterparts
-        if isinstance(data.dtype, BaseMaskedDtype) and data.dtype in pdoptional_to_numpy_typemap:
-            data = data.astype(pdoptional_to_numpy_typemap[data.dtype])
-        if data.dtype.kind == "O" and not is_categorical_dtype(data.dtype):
-            try:
-                if selement.type in [parquet_thrift.Type.INT64,
-                                     parquet_thrift.Type.INT32]:
-                    data = data.astype(int)
-                elif selement.type == parquet_thrift.Type.BOOLEAN:
-                    data = data.astype(bool)
-            except ValueError as e:
-                t = parquet_thrift.Type._VALUES_TO_NAMES[selement.type]
-                raise ValueError('Error converting column "%s" to primitive '
-                                 'type %s. Original error: '
-                                 '%s' % (data.name, t, e))
-
-    else:
-        definition_data = b""
-        num_nulls = 0
-    num_nulls = int(num_nulls)
-
-    # No nested field handling (encode those as J/BSON)
-    repetition_data = b""
-
+    first_page = True
     cats = False
-    name = data.name
+    name = data0.name
     diff = 0
     max, min = None, None
-    start = f.tell()
-
-    if is_categorical_dtype(data.dtype):
-        dph = parquet_thrift.DictionaryPageHeader(
-            num_values=check_32(len(data.cat.categories)),
-            encoding=parquet_thrift.Encoding.PLAIN,
-            i32=1
-        )
-        bdata = encode['PLAIN'](pd.Series(data.cat.categories), selement)
-        l0 = len(bdata)
-        if compression and compression.upper() != "UNCOMPRESSED":
-            bdata = compress_data(bdata, compression)
-            l1 = len(bdata)
-        else:
-            l1 = l0
-        diff += l0 - l1
-        ph = parquet_thrift.PageHeader(
-                type=parquet_thrift.PageType.DICTIONARY_PAGE,
-                uncompressed_page_size=check_32(l0),
-                compressed_page_size=check_32(l1),
-                dictionary_page_header=dph, crc=None, i32=1)
-
-        dict_start = f.tell()
-        write_thrift(f, ph)
-        f.write(bdata)
+    column_chunk_start = f.tell()
+    rows_per_page = _rows_per_page(data0, selement, has_nulls)
+    row_offsets = list(range(0, len(data0), rows_per_page)) + [len(data0)]
+    global_num_nulls = 0
+    dict_page_offset = None
+    data_page_offset = column_chunk_start
+
+    # column global stats
+    if is_categorical_dtype(data0.dtype) and stats:
         try:
-            if stats:
-                # TODO: this max/min works, but is slow
-                max, min = np.array(data[data.notnull()]).max(), np.array(data[data.notnull()]).min()
-                if pd.isna(max):
-                    stats = False
+            dnnu = data0.unique().as_ordered()
+            max, min = dnnu.max(), dnnu.min()
+            if pd.isna(max):
+                stats = False
+            else:
+                if selement.type == parquet_thrift.Type.BYTE_ARRAY:
+                    if selement.converted_type is not None:
+                        max = encode['PLAIN'](pd.Series([max]), selement)[4:]
+                        min = encode['PLAIN'](pd.Series([min]), selement)[4:]
                 else:
-                    if selement.type == parquet_thrift.Type.BYTE_ARRAY:
-                        if selement.converted_type is not None:
-                            max = encode['PLAIN'](pd.Series([max]), selement)[4:]
-                            min = encode['PLAIN'](pd.Series([min]), selement)[4:]
-                    else:
-                        max = encode['PLAIN'](pd.Series([max]), selement)
-                        min = encode['PLAIN'](pd.Series([min]), selement)
+                    max = encode['PLAIN'](pd.Series([max]), selement)
+                    min = encode['PLAIN'](pd.Series([min]), selement)
         except (TypeError, ValueError):
             stats = False
-        ncats = len(data.cat.categories)
-        dcat = data.cat.categories.dtype
-        data = data.cat.codes
-        cats = True
-        encoding = "RLE_DICTIONARY"
-    elif str(data.dtype) in ['int8', 'int16', 'uint8', 'uint16']:
-        # encoding = "RLE"
-        # disallow bit-packing for compatibility
-        data = data.astype('int32')
-
-    try:
-        if encoding != 'RLE_DICTIONARY':
-            # for categorical, we already did this above
-            if stats:
-                max, min = data[data.notnull()].values.max(), data[data.notnull()].values.min()
-                if pd.isna(max):
-                    stats = False
+    elif stats:
+        try:
+            max, min = data0.max(), data0.min()
+            if pd.isna(max):
+                stats = False
+            else:
+                if selement.type == parquet_thrift.Type.BYTE_ARRAY:
+                    if selement.converted_type is not None:
+                        # max = max.encode("utf8") ?
+                        max = encode['PLAIN'](pd.Series([max], name=name), selement)[4:]
+                        min = encode['PLAIN'](pd.Series([min], name=name), selement)[4:]
                 else:
-                    if selement.type == parquet_thrift.Type.BYTE_ARRAY:
-                        if selement.converted_type is not None:
-                            # max = max.encode("utf8") ?
-                            max = encode['PLAIN'](pd.Series([max], name=data.name), selement)[4:]
-                            min = encode['PLAIN'](pd.Series([min], name=data.name), selement)[4:]
-                    else:
-                        max = encode['PLAIN'](pd.Series([max], name=data.name, dtype=data.dtype), selement)
-                        min = encode['PLAIN'](pd.Series([min], name=data.name, dtype=data.dtype), selement)
-    except (TypeError, ValueError):
-        stats = False
-    s = parquet_thrift.Statistics(max=max, min=min, null_count=num_nulls) if stats else None
-
-    if datapage_version == 1:
-        bdata = b"".join([
-            repetition_data, definition_data, encode[encoding](data, selement), 8 * b'\x00'
-        ])
-        dph = parquet_thrift.DataPageHeader(
-            num_values=check_32(tot_rows),
-            encoding=getattr(parquet_thrift.Encoding, encoding),
-            definition_level_encoding=parquet_thrift.Encoding.RLE,
-            repetition_level_encoding=parquet_thrift.Encoding.BIT_PACKED,
-            i32=1
-        )
-        l0 = len(bdata)
-
-        if compression:
-            bdata = compress_data(bdata, compression)
-            l1 = len(bdata)
-        else:
-            l1 = l0
-        diff += l0 - l1
-
-        ph = parquet_thrift.PageHeader(type=parquet_thrift.PageType.DATA_PAGE,
-                                       uncompressed_page_size=check_32(l0),
-                                       compressed_page_size=check_32(l1),
-                                       data_page_header=dph, i32=1)
-        write_thrift(f, ph)
-        f.write(bdata)
-    elif datapage_version == 2:
-        is_compressed = isinstance(compression, dict) or (
-            compression is not None and compression.upper() != "UNCOMPRESSED")
-        dph = parquet_thrift.DataPageHeaderV2(
-            num_values=check_32(tot_rows),
-            num_nulls=check_32(num_nulls),
-            num_rows=check_32(tot_rows),
-            encoding=getattr(parquet_thrift.Encoding, encoding),
-            definition_levels_byte_length=len(definition_data),
-            repetition_levels_byte_length=0,  # len(repetition_data),
-            is_compressed=is_compressed,
-            statistics=s,
-            i32=1
-        )
-        bdata = encode[encoding](data, selement)
-        lb = len(bdata)
-        if is_compressed:
-            bdata = compress_data(bdata, compression)
-            diff = lb - len(bdata)
+                    max = encode['PLAIN'](pd.Series([max], name=name, dtype=data0.dtype), selement)
+                    min = encode['PLAIN'](pd.Series([min], name=name, dtype=data0.dtype), selement)
+        except (TypeError, ValueError):
+            stats = False
+
+    for row_start, row_end in zip(row_offsets[:-1], row_offsets[1:]):
+        data = data0.iloc[row_start:row_end]
+        if has_nulls:
+            if is_categorical_dtype(data.dtype):
+                num_nulls = (data.cat.codes == -1).sum()
+            else:
+                num_nulls = len(data) - data.count()
+            definition_data, data = make_definitions(data, num_nulls == 0, datapage_version=datapage_version)
+            # make_definitions returns `data` with all nulls dropped
+            # the null-stripped `data` can be converted from Optional Types to
+            # their numpy counterparts
+            if isinstance(data.dtype, BaseMaskedDtype) and data.dtype in pdoptional_to_numpy_typemap:
+                data = data.astype(pdoptional_to_numpy_typemap[data.dtype], copy=False)
+            if data.dtype.kind == "O" and not is_categorical_dtype(data.dtype):
+                try:
+                    if selement.type == parquet_thrift.Type.INT64:
+                        data = data.astype("int64", copy=False)
+                    elif selement.type == parquet_thrift.Type.INT32:
+                        data = data.astype("int32", copy=False)
+                    elif selement.type == parquet_thrift.Type.BOOLEAN:
+                        data = data.astype(bool, copy=False)
+                except ValueError as e:
+                    t = parquet_thrift.Type._VALUES_TO_NAMES[selement.type]
+                    raise ValueError('Error converting column "%s" to primitive '
+                                     'type %s. Original error: '
+                                     '%s' % (data.name, t, e))
         else:
-            diff = 0
-        ph = parquet_thrift.PageHeader(type=parquet_thrift.PageType.DATA_PAGE_V2,
-                                       uncompressed_page_size=check_32(lb + len(definition_data)),
-                                       compressed_page_size=check_32(len(bdata) + len(definition_data)),
-                                       data_page_header_v2=dph, i32=1)
-        write_thrift(f, ph)
-        # f.write(repetition_data)  # no-op
-        f.write(definition_data)
-        f.write(bdata)
-
-    compressed_size = f.tell() - start
-    uncompressed_size = compressed_size + diff
+            definition_data = b""
+            num_nulls = 0
+        num_nulls = int(num_nulls)
+        global_num_nulls += num_nulls
+
+        # No nested field handling (encode those as J/BSON)
+        repetition_data = b""
+
+        if is_categorical_dtype(data.dtype):
+            if first_page:
+                # make "index page"
+                dict_page_offset = column_chunk_start
+                dph = parquet_thrift.DictionaryPageHeader(
+                    num_values=check_32(len(data.cat.categories)),
+                    encoding=parquet_thrift.Encoding.PLAIN,
+                    i32=1
+                )
+                bdata = encode['PLAIN'](pd.Series(data.cat.categories), selement)
+                l0 = len(bdata)
+                if compression and compression.upper() != "UNCOMPRESSED":
+                    bdata = compress_data(bdata, compression)
+                    l1 = len(bdata)
+                else:
+                    l1 = l0
+                diff += l0 - l1
+                ph = parquet_thrift.PageHeader(
+                        type=parquet_thrift.PageType.DICTIONARY_PAGE,
+                        uncompressed_page_size=check_32(l0),
+                        compressed_page_size=check_32(l1),
+                        dictionary_page_header=dph, crc=None, i32=1)
+
+                write_thrift(f, ph)
+                f.write(bdata)
+                data_page_offset = f.tell()
+                ncats = len(data.cat.categories)
+                dcat = data.cat.categories.dtype
+                cats = True
+                encoding = "RLE_DICTIONARY"
+            data = data.cat.codes
+        if str(data0.dtype) in ['int8', 'int16', 'uint8', 'uint16']:
+            # PLAIN encoding must be upcast to parquet primitive
+            data = data.astype('int32')
+
+        if datapage_version == 1:
+            bdata = b"".join([
+                repetition_data, definition_data, encode[encoding](data, selement), 8 * b'\x00'
+            ])
+            dph = parquet_thrift.DataPageHeader(
+                num_values=check_32(row_end - row_start),
+                encoding=getattr(parquet_thrift.Encoding, encoding),
+                definition_level_encoding=parquet_thrift.Encoding.RLE,
+                repetition_level_encoding=parquet_thrift.Encoding.BIT_PACKED,
+                i32=1
+            )
+            l0 = len(bdata)
 
-    offset = f.tell()
+            if compression:
+                bdata = compress_data(bdata, compression)
+                l1 = len(bdata)
+            else:
+                l1 = l0
+            diff += l0 - l1
+
+            ph = parquet_thrift.PageHeader(type=parquet_thrift.PageType.DATA_PAGE,
+                                           uncompressed_page_size=check_32(l0),
+                                           compressed_page_size=check_32(l1),
+                                           data_page_header=dph, i32=1)
+            write_thrift(f, ph)
+            f.write(bdata)
+        elif datapage_version == 2:
+            is_compressed = isinstance(compression, dict) or (
+                compression is not None and compression.upper() != "UNCOMPRESSED")
+            dph = parquet_thrift.DataPageHeaderV2(
+                num_values=check_32(row_end - row_start),
+                num_nulls=check_32(num_nulls),
+                num_rows=check_32(row_end - row_start),
+                encoding=getattr(parquet_thrift.Encoding, encoding),
+                definition_levels_byte_length=len(definition_data),
+                repetition_levels_byte_length=0,  # len(repetition_data),
+                is_compressed=is_compressed,
+                statistics=None,
+                i32=1
+            )
+            bdata = encode[encoding](data, selement)
+            lb = len(bdata)
+            if is_compressed:
+                bdata = compress_data(bdata, compression)
+                diff += lb - len(bdata)
+            else:
+                diff += 0
+            ph = parquet_thrift.PageHeader(
+                type=parquet_thrift.PageType.DATA_PAGE_V2,
+                uncompressed_page_size=check_32(lb + len(definition_data)),
+                compressed_page_size=check_32(len(bdata) + len(definition_data)),
+                data_page_header_v2=dph, i32=1)
+            write_thrift(f, ph)
+            # f.write(repetition_data)  # no-op
+            f.write(definition_data)
+            f.write(bdata)
+        first_page = False
+
+    compressed_size = f.tell() - column_chunk_start
+    uncompressed_size = compressed_size + diff
 
+    # encoding stats for thrift metadata
     if cats:
         p = [
             parquet_thrift.PageEncodingStats(
@@ -628,7 +676,8 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
                 encoding=parquet_thrift.Encoding.PLAIN, count=1, i32=1),
             parquet_thrift.PageEncodingStats(
                 page_type=parquet_thrift.PageType.DATA_PAGE,
-                encoding=parquet_thrift.Encoding.RLE_DICTIONARY, count=1, i32=1),
+                encoding=parquet_thrift.Encoding.RLE_DICTIONARY,
+                count=len(row_offsets) - 1, i32=1),
         ]
         encodings = [parquet_thrift.Encoding.PLAIN,
                      parquet_thrift.Encoding.RLE_DICTIONARY]
@@ -636,7 +685,8 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
     else:
         p = [parquet_thrift.PageEncodingStats(
              page_type=parquet_thrift.PageType.DATA_PAGE,
-             encoding=parquet_thrift.Encoding.PLAIN, count=1, i32=1)]
+             encoding=parquet_thrift.Encoding.PLAIN,
+             count=len(row_offsets) - 1, i32=1)]
         encodings = [parquet_thrift.Encoding.PLAIN]
 
     if isinstance(compression, dict):
@@ -644,6 +694,9 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
     else:
         algorithm = compression
 
+    # output thrift metadata
+    s = parquet_thrift.Statistics(max=max, min=min, null_count=global_num_nulls) if stats else None
+
     kvm = []
     if isinstance(name, (list, tuple)):
         name = str(tuple(name))
@@ -655,7 +708,8 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
                if algorithm else 0),
         num_values=tot_rows,
         statistics=s,
-        data_page_offset=start,
+        data_page_offset=data_page_offset,
+        dictionary_page_offset=dict_page_offset,
         encoding_stats=p,
         key_value_metadata=kvm,
         total_uncompressed_size=uncompressed_size,
@@ -663,14 +717,13 @@ def write_column(f, data, selement, compression=None, datapage_version=None,
         i32list=[1, 4]
     )
     if cats:
-        cmd.dictionary_page_offset = dict_start
         kvm.append(
             parquet_thrift.KeyValue(key='num_categories', value=str(ncats)))
         kvm.append(
             parquet_thrift.KeyValue(key='numpy_dtype', value=str(data.dtype)))
         kvm.append(
             parquet_thrift.KeyValue(key='label_dtype', value=str(dcat)))
-    chunk = parquet_thrift.ColumnChunk(file_offset=offset,
+    chunk = parquet_thrift.ColumnChunk(file_offset=column_chunk_start,
                                        meta_data=cmd,
                                        file_path=None)
     return chunk
@@ -680,31 +733,9 @@ class DataFrameSizeWarning(UserWarning):
     pass
 
 
-WARNING_THRESHOLD = 2**30
-
-
-def warn_size(df, limit=None):
-    limit = limit or WARNING_THRESHOLD
-    total = 0
-    rows = len(df)
-    for col, dtype in df.dtypes.items():
-        if dtype.kind in ['f', 'i', 'M', 'm']:
-            # ignores nullability here, since the bools don't take up much space
-            total += dtype.itemsize * rows
-
-        else:
-            approx = sum([len(str(_)) + 4 for _ in df[col].iloc[:100]])
-            total += (approx * rows) // 100
-    if total > limit:
-        warnings.warn(DataFrameSizeWarning(
-            f"Parquet partition about warning size limit, {total} > {limit}"
-        ))
-
-
 def make_row_group(f, data, schema, compression=None, stats=True):
     """ Make a single row group of a Parquet file """
     rows = len(data)
-    warn_size(data)
     if rows == 0:
         return
     if isinstance(data.columns, pd.MultiIndex):
@@ -739,7 +770,7 @@ def make_row_group(f, data, schema, compression=None, stats=True):
             if isinstance(stats, int):
                 st = stats
             elif stats == "auto":
-                st = coldata.dtype.kind in ["i", "f", "M"]
+                st = coldata.dtype.kind in ["i", "u", "f", "M"]
             else:
                 st = column.name in stats
             chunk = write_column(f, coldata, column,
@@ -1071,7 +1102,7 @@ def write(filename, data, row_group_offsets=None,
           mkdirs=None, has_nulls=True, write_index=None,
           partition_on=[], fixed_text=None, append=False,
           object_encoding='infer', times='int64',
-          custom_metadata=None, stats=True):
+          custom_metadata=None, stats="auto"):
     """Write pandas dataframe to filename with parquet format.
 
     Parameters
@@ -1189,11 +1220,10 @@ def write(filename, data, row_group_offsets=None,
         Ignored if appending to an existing parquet data-set.
     stats: True|False|list(str)|"auto"
         Whether to calculate and write summary statistics.
-        If True (default), do it for every column;
+        If True, do it for every column;
         If False, never do;
         And if a list of str, do it only for those specified columns.
-        "auto" means True for any int/float or timemstamp column, False
-        otherwise. This will become the default in a future release.
+        "auto" (default) means True for any int/float or timestamp column
 
     Examples
     --------
@@ -1238,7 +1268,7 @@ def write(filename, data, row_group_offsets=None,
         # Case 'append=False'.
         # Define 'index_cols' to be recorded in metadata.
         if (write_index or write_index is None
-            and not isinstance(data.index, pd.RangeIndex)):
+                and not isinstance(data.index, pd.RangeIndex)):
             # Keep name(s) of index to metadata.
             cols = set(data)
             data = reset_row_idx(data)


Test patch:
diff --git a/fastparquet/test/test_api.py b/fastparquet/test/test_api.py
index 42176c8..bd69278 100644
--- a/fastparquet/test/test_api.py
+++ b/fastparquet/test/test_api.py
@@ -39,7 +39,7 @@ def test_statistics(tempdir):
                        'z': ['a', 'b', 'c']})
 
     fn = os.path.join(tempdir, 'foo.parquet')
-    write(fn, df, row_group_offsets=[0, 2])
+    write(fn, df, row_group_offsets=[0, 2], stats=True)
 
     p = ParquetFile(fn)
 
@@ -111,8 +111,9 @@ def test_sorted_row_group_columns(tempdir):
                        'z': ['a', 'b', 'c', 'd']})
 
     fn = os.path.join(tempdir, 'foo.parquet')
-    write(fn, df, row_group_offsets=[0, 2], object_encoding={'v': 'json',
-                                                             'z': 'utf8'})
+    write(fn, df, row_group_offsets=[0, 2],
+          object_encoding={'v': 'json', 'z': 'utf8'},
+          stats=True)
 
     pf = ParquetFile(fn)
 
@@ -316,6 +317,8 @@ def test_filelike(tempdir):
 
 
 def test_cast_index(tempdir):
+    if pd.__version__.split(".", 1)[0] > "1":
+        pytest.skip()
     df = pd.DataFrame({'i8': np.array([1, 2, 3, 4], dtype='uint8'),
                        'i16': np.array([1, 2, 3, 4], dtype='int16'),
                        'i32': np.array([1, 2, 3, 4], dtype='int32'),
diff --git a/fastparquet/test/test_output.py b/fastparquet/test/test_output.py
index 645e53f..90f718e 100644
--- a/fastparquet/test/test_output.py
+++ b/fastparquet/test/test_output.py
@@ -500,7 +500,7 @@ def test_text_convert(tempdir):
                        'b': [b'a'] * 100})
     fn = os.path.join(tempdir, 'tmp.parq')
 
-    write(fn, df, fixed_text={'a': 2, 'b': 1})
+    write(fn, df, fixed_text={'a': 2, 'b': 1}, stats=True)
     pf = ParquetFile(fn)
     assert pf._schema[1].type == parquet_thrift.Type.FIXED_LEN_BYTE_ARRAY
     assert pf._schema[1].type_length == 2
@@ -510,7 +510,7 @@ def test_text_convert(tempdir):
     df2 = pf.to_pandas()
     tm.assert_frame_equal(df, df2, check_categorical=False)
 
-    write(fn, df)
+    write(fn, df, stats=True)
     pf = ParquetFile(fn)
     assert pf._schema[1].type == parquet_thrift.Type.BYTE_ARRAY
     assert pf._schema[2].type == parquet_thrift.Type.BYTE_ARRAY
@@ -518,7 +518,7 @@ def test_text_convert(tempdir):
     df2 = pf.to_pandas()
     tm.assert_frame_equal(df, df2, check_categorical=False)
 
-    write(fn, df, fixed_text={'a': 2})
+    write(fn, df, fixed_text={'a': 2}, stats=True)
     pf = ParquetFile(fn)
     assert pf._schema[1].type == parquet_thrift.Type.FIXED_LEN_BYTE_ARRAY
     assert pf._schema[2].type == parquet_thrift.Type.BYTE_ARRAY
@@ -548,10 +548,7 @@ def test_null_time(tempdir):
     assert sum(data['t'].isnull()) == sum(expected['t'].isnull())
 
 
-@pytest.mark.parametrize(
-    "pnull", [True, False]
-)
-def test_auto_null_object(tempdir, pnull):
+def test_auto_null_object(tempdir):
     tmp = str(tempdir)
     df = pd.DataFrame({'a': [1, 2, 3, 0],
                        'aa': pd.Series([1, 2, 3, None], dtype=object),
@@ -571,24 +568,6 @@ def test_auto_null_object(tempdir, pnull):
         write(fn, df, has_nulls=False)
 
     write(fn, df, has_nulls=True)
-    pf = ParquetFile(fn, pandas_nulls=pnull)
-    for col in pf._schema[1:]:
-        assert col.repetition_type == parquet_thrift.FieldRepetitionType.OPTIONAL
-    df2 = pf.to_pandas(categories=['e'])
-
-    tm.assert_frame_equal(df[test_cols], df2[test_cols], check_categorical=False,
-                          check_dtype=False)
-    tm.assert_frame_equal(df[['bb']].astype('float64'), df2[['bb']])
-    tm.assert_frame_equal(df[['aaa']].astype('int64'), df2[['aaa']])
-    if pnull:
-        tm.assert_frame_equal(df[['aa']].astype('Int64'), df2[['aa']])
-        tm.assert_frame_equal(df[['ff']].astype("boolean"), df2[['ff']])
-    else:
-        tm.assert_frame_equal(df[['aa']].astype('float'), df2[['aa']])
-        tm.assert_frame_equal(df[['ff']].astype("float"), df2[['ff']])
-
-    # not giving any value same as has_nulls=True
-    write(fn, df)
     pf = ParquetFile(fn)
     for col in pf._schema[1:]:
         assert col.repetition_type == parquet_thrift.FieldRepetitionType.OPTIONAL
@@ -596,11 +575,12 @@ def test_auto_null_object(tempdir, pnull):
 
     tm.assert_frame_equal(df[test_cols], df2[test_cols], check_categorical=False,
                           check_dtype=False)
-    tm.assert_frame_equal(df[['ff']].astype('boolean'), df2[['ff']])
     tm.assert_frame_equal(df[['bb']].astype('float64'), df2[['bb']])
-    tm.assert_frame_equal(df[['aaa']].astype('int64'), df2[['aaa']])
+    tm.assert_frame_equal(df[['aa']].astype('Int64'), df2[['aa']])
+    tm.assert_frame_equal(df[['ff']].astype("boolean"), df2[['ff']])
+    tm.assert_frame_equal(df[['aaa']].astype('Int64'), df2[['aaa']])
 
-    # 'infer' is new recommended auto-null
+    # 'infer' is equivalent of None, previous default
     write(fn, df, has_nulls='infer')
     pf = ParquetFile(fn)
     for col in pf._schema[1:]:
@@ -612,21 +592,7 @@ def test_auto_null_object(tempdir, pnull):
     tm.assert_frame_equal(df[test_cols], df2[test_cols], check_categorical=False)
     tm.assert_frame_equal(df[['ff']].astype('boolean'), df2[['ff']])
     tm.assert_frame_equal(df[['bb']].astype('float64'), df2[['bb']])
-    tm.assert_frame_equal(df[['aaa']].astype('int64'), df2[['aaa']])
-
-    # nut legacy None still works
-    write(fn, df, has_nulls=None)
-    pf = ParquetFile(fn)
-    for col in pf._schema[1:]:
-        if col.name in object_cols:
-            assert col.repetition_type == parquet_thrift.FieldRepetitionType.OPTIONAL
-        else:
-            assert col.repetition_type == parquet_thrift.FieldRepetitionType.REQUIRED
-    df2 = pf.to_pandas()
-    tm.assert_frame_equal(df[test_cols], df2[test_cols], check_categorical=False)
-    tm.assert_frame_equal(df[['ff']].astype('boolean'), df2[['ff']])
-    tm.assert_frame_equal(df[['bb']].astype('float64'), df2[['bb']])
-    tm.assert_frame_equal(df[['aaa']].astype('int64'), df2[['aaa']])
+    tm.assert_frame_equal(df[['aaa']].astype('Int64'), df2[['aaa']])
 
 
 @pytest.mark.parametrize('n', (10, 127, 2**8 + 1, 2**16 + 1))
@@ -1030,14 +996,6 @@ def test_float(tempdir):
     assert (out.v == df.v).all()
 
 
-def test_limit(monkeypatch, tempdir):
-    fn = os.path.join(tempdir, 'temp.parq')
-    monkeypatch.setattr(writer, "WARNING_THRESHOLD", 1)
-    df = pd.DataFrame({'a': [0]})
-    with pytest.warns(writer.DataFrameSizeWarning):
-        write(fn, df)
-
-
 def test_empty_columns(tempdir):
     fn = os.path.join(tempdir, 'temp.parq')
     df = pd.DataFrame(
@@ -1198,4 +1156,44 @@ def test_roundtrip_pathlib_path(tempdir):
     file_path = Path(tempdir) / "out.parq"
     write(file_path, df)
     out = ParquetFile(file_path).to_pandas()
-    assert out.to_dict() == df.to_dict()
\ No newline at end of file
+    assert out.to_dict() == df.to_dict()
+
+
+def test_pagesize(monkeypatch, tempdir):
+    monkeypatch.setattr(writer, "MAX_PAGE_SIZE", 50)
+    fn = os.path.join(tempdir, "out.parq")
+    # 480 bytes + 60/8 because optional -> 10 pages
+    df = pd.DataFrame({'a': [0, 1, 2] * 20}, dtype="int64")
+    write(fn, df)
+    pf = ParquetFile(fn)
+    col = pf.row_groups[0].columns[0]
+    assert col.meta_data.dictionary_page_offset is None
+    assert col.file_offset == 4
+    assert col.meta_data.statistics.max == b'\x02\x00\x00\x00\x00\x00\x00\x00'
+    assert col.meta_data.data_page_offset == 4
+    enc = col.meta_data.encoding_stats
+    assert len(enc) == 1
+    assert enc[0].count == 10
+    out = pf.to_pandas()
+    assert out.to_dict() == df.to_dict()
+
+
+def test_pagesize_cat(monkeypatch, tempdir):
+    monkeypatch.setattr(writer, "MAX_PAGE_SIZE", 10)
+    fn = os.path.join(tempdir, "out.parq")
+    # 60 bytes + 60/8 because optional -> 8 pages
+    df = pd.DataFrame({'a': ["0", "1", "2"] * 20}, dtype="category")
+    write(fn, df)
+    pf = ParquetFile(fn)
+    col = pf.row_groups[0].columns[0]
+    assert col.meta_data.dictionary_page_offset == 4
+    assert col.file_offset == 4
+    assert col.meta_data.statistics is None
+    assert col.meta_data.data_page_offset > 4
+    enc = col.meta_data.encoding_stats
+    assert len(enc) == 2
+    assert enc[0].count == 1  # dict comes first
+    assert enc[1].count == 8  # dict comes first
+    out = pf.to_pandas()
+    assert out.to_dict() == df.to_dict()
+
diff --git a/fastparquet/test/test_read.py b/fastparquet/test/test_read.py
index 944ac4c..5f8139f 100644
--- a/fastparquet/test/test_read.py
+++ b/fastparquet/test/test_read.py
@@ -252,7 +252,7 @@ def test_statistics(tempdir):
     s = pd.Series([b'a', b'b', b'c']*20)
     df = pd.DataFrame({'a': s, 'b': s.astype('category'),
                        'c': s.astype('category').cat.as_ordered()})
-    fastparquet.write(tempdir, df, file_scheme='hive')
+    fastparquet.write(tempdir, df, file_scheme='hive', stats=True)
     pf = fastparquet.ParquetFile(tempdir)
     stat = pf.statistics
     assert stat['max']['a'] == [b'c']