def METHOD_NAME(
    self,
    table: str,
    row_count_threshold: int,
    sa_connection: "sqlalchemy.engine.base.Connection",
    custom_partition_column: Optional[str],
) -> str:
    """Return a dictionary defining how to partition the Spark SQL extraction."""
    row_count = self._get_table_count_from_stats(table, sa_connection=sa_connection)
    if not row_count:
        # In case this is a new table with no stats, do a full count
        row_count = self._get_table_count(table, sa_connection=sa_connection)
    if row_count < int(row_count_threshold):
        # The table does not have enough rows to merit partitioning Spark SQL read.
        return None
    accepted_data_types = ["int", "bigint", "mediumint"]
    if custom_partition_column:
        # The user provided a partition column.
        column = self._normalise_column_name(
            table, custom_partition_column, sa_connection
        )
        if not column:
            return {
                PARTITION_COMMENT: f"Serial read, column does not exist: {custom_partition_column}"
            }
        partition_options = self._define_native_column_read_partitioning(
            table,
            column,
            accepted_data_types,
            row_count,
            row_count_threshold,
            "user provided column",
            sa_connection,
        )
        if partition_options:
            return partition_options
    pk_cols = self.get_primary_keys().get(table)
    if pk_cols and len(pk_cols) == 1:
        # Partition by primary key singleton.
        column = pk_cols[0]
        partition_options = self._define_native_column_read_partitioning(
            table,
            column,
            accepted_data_types,
            row_count,
            row_count_threshold,
            "primary key column",
            sa_connection,
        )
        if partition_options:
            return partition_options
    return None