Add global default

Author: nalimilan

docs/src/lib/functions.md

@@ -129,3 +129,9 @@ pairs
 ```@docs
 isapprox
 ```
+
+## Multithreading
+```@docs
+DataFrames.nthreads
+DataFrames.nthreads!
+```

src/abstractdataframe/selection.jl

@@ -576,8 +576,12 @@ end
 """
     select!(df::DataFrame, args...; renamecols::Bool=true)
     select!(args::Base.Callable, df::DataFrame; renamecols::Bool=true)
-    select!(gd::GroupedDataFrame{DataFrame}, args...; ungroup::Bool=true, renamecols::Bool=true)
-    select!(f::Base.Callable, gd::GroupedDataFrame; ungroup::Bool=true, renamecols::Bool=true)
+    select!(gd::GroupedDataFrame{DataFrame}, args...;
+            ungroup::Bool=true, renamecols::Bool=true,
+            nthreads::Int=DataFrames.nthreads())
+    select!(f::Base.Callable, gd::GroupedDataFrame;
+            ungroup::Bool=true, renamecols::Bool=true,
+            nthreads::Int=DataFrames.nthreads())
 
 Mutate `df` or `gd` in place to retain only columns or transformations specified by `args...` and
 return it. The result is guaranteed to have the same number of rows as `df` or
@@ -595,6 +599,11 @@ $TRANSFORMATION_COMMON_RULES
   column names should include the name of transformation functions or not.
 - `ungroup::Bool=true` : whether the return value of the operation on `gd` should be a data
   frame or a `GroupedDataFrame`.
+- `nthreads::Int=DataFrames.nthreads()` : the number of CPU threads to use.
+  Defaults to `1` unless [`DataFrames.nthreads!`](@ref) has been called with
+  a different value. Passing a value higher than 1 currently has an effect only
+  for some optimized grouped reductions. Values higher than `Threads.nthreads()`
+  will be replaced with that value.
 
 See [`select`](@ref) for examples.
 ```
@@ -613,8 +622,12 @@ end
 """
     transform!(df::DataFrame, args...; renamecols::Bool=true)
     transform!(args::Callable, df::DataFrame; renamecols::Bool=true)
-    transform!(gd::GroupedDataFrame{DataFrame}, args...; ungroup::Bool=true, renamecols::Bool=true)
-    transform!(f::Base.Callable, gd::GroupedDataFrame; ungroup::Bool=true, renamecols::Bool=true)
+    transform!(gd::GroupedDataFrame{DataFrame}, args...;
+               ungroup::Bool=true, renamecols::Bool=true,
+               nthreads::Int=DataFrames.nthreads())
+    transform!(f::Base.Callable, gd::GroupedDataFrame;
+               ungroup::Bool=true, renamecols::Bool=true,
+               nthreads::Int=DataFrames.nthreads())
 
 Mutate `df` or `gd` in place to add columns specified by `args...` and return it.
 The result is guaranteed to have the same number of rows as `df`.
@@ -627,6 +640,11 @@ $TRANSFORMATION_COMMON_RULES
   column names should include the name of transformation functions or not.
 - `ungroup::Bool=true` : whether the return value of the operation on `gd` should be a data
   frame or a `GroupedDataFrame`.
+- `nthreads::Int=DataFrames.nthreads()` : the number of CPU threads to use.
+  Defaults to `1` unless [`DataFrames.nthreads!`](@ref) has been called with
+  a different value. Passing a value higher than 1 currently has an effect only
+  for some optimized grouped reductions. Values higher than `Threads.nthreads()`
+  will be replaced with that value.
 
 See [`select`](@ref) for examples.
 """
@@ -644,10 +662,10 @@ end
     select(df::AbstractDataFrame, args...; copycols::Bool=true, renamecols::Bool=true)
     select(args::Callable, df::DataFrame; renamecols::Bool=true)
     select(gd::GroupedDataFrame, args...; copycols::Bool=true, keepkeys::Bool=true,
-           ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=1)
+           ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=DataFrames.nthreads())
     select(f::Base.Callable, gd::GroupedDataFrame; copycols::Bool=true,
            keepkeys::Bool=true, ungroup::Bool=true,
-           renamecols::Bool=true, nthreads::Int=1)
+           renamecols::Bool=true, nthreads::Int=DataFrames.nthreads())
 
 Create a new data frame that contains columns from `df` or `gd` specified by
 `args` and return it. The result is guaranteed to have the same number of rows
@@ -665,9 +683,11 @@ $TRANSFORMATION_COMMON_RULES
   data frame.
 - `ungroup::Bool=true` : whether the return value of the operation on `gd` should be a data
   frame or a `GroupedDataFrame`.
-- `nthreads::Int=1` : the number of CPU threads to use. Passing a value higher than 1
-  currently has an effect only for some optimized grouped reductions. Values higher than
-  `Threads.nthreads()` will be replaced with that value.
+- `nthreads::Int=DataFrames.nthreads()` : the number of CPU threads to use.
+  Defaults to `1` unless [`DataFrames.nthreads!`](@ref) has been called with
+  a different value. Passing a value higher than 1 currently has an effect only
+  for some optimized grouped reductions. Values higher than `Threads.nthreads()`
+  will be replaced with that value.
 
 # Examples
 ```jldoctest
@@ -863,10 +883,10 @@ end
     transform(f::Callable, df::DataFrame; renamecols::Bool=true)
     transform(gd::GroupedDataFrame, args...; copycols::Bool=true,
               keepkeys::Bool=true, ungroup::Bool=true,
-              renamecols::Bool=true, nthreads::Int=1)
+              renamecols::Bool=true, nthreads::Int=DataFrames.nthreads())
     transform(f::Base.Callable, gd::GroupedDataFrame; copycols::Bool=true,
               keepkeys::Bool=true, ungroup::Bool=true,
-              renamecols::Bool=true, nthreads::Int=1)
+              renamecols::Bool=true, nthreads::Int=DataFrames.nthreads())
 
 Create a new data frame that contains columns from `df` or `gd` plus columns
 specified by `args` and return it. The result is guaranteed to have the same
@@ -883,9 +903,11 @@ $TRANSFORMATION_COMMON_RULES
   data frame.
 - `ungroup::Bool=true` : whether the return value of the operation on `gd` should be a data
   frame or a `GroupedDataFrame`.
-- `nthreads::Int=1` : the number of CPU threads to use. Passing a value higher than 1
-  currently has an effect only for some optimized grouped reductions. Values higher than
-  `Threads.nthreads()` will be replaced with that value.
+- `nthreads::Int=DataFrames.nthreads()` : the number of CPU threads to use.
+  Defaults to `1` unless [`DataFrames.nthreads!`](@ref) has been called with
+  a different value. Passing a value higher than 1 currently has an effect only
+  for some optimized grouped reductions. Values higher than `Threads.nthreads()`
+  will be replaced with that value.
 
 Note that when the first argument is a `GroupedDataFrame`, `keepkeys=false`
 is needed to be able to return a different value for the grouping column:
@@ -934,10 +956,10 @@ end
     combine(f::Callable, df::AbstractDataFrame; renamecols::Bool=true)
     combine(gd::GroupedDataFrame, args...;
             keepkeys::Bool=true, ungroup::Bool=true,
-            renamecols::Bool=true, nthreads::Int=1)
+            renamecols::Bool=true, nthreads::Int=DataFrames.nthreads())
     combine(f::Base.Callable, gd::GroupedDataFrame;
             keepkeys::Bool=true, ungroup::Bool=true,
-            renamecols::Bool=true, nthreads::Int=1)
+            renamecols::Bool=true, nthreads::Int=DataFrames.nthreads())
 
 Create a new data frame that contains columns from `df` or `gd` specified by
 `args` and return it. The result can have any number of rows that is determined
@@ -952,9 +974,11 @@ $TRANSFORMATION_COMMON_RULES
   data frame.
 - `ungroup::Bool=true` : whether the return value of the operation on `gd` should be a data
   frame or a `GroupedDataFrame`.
-- `nthreads::Int=1` : the number of CPU threads to use. Passing a value higher than 1
-  currently has an effect only for some optimized grouped reductions. Values higher than
-  `Threads.nthreads()` will be replaced with that value.
+- `nthreads::Int=DataFrames.nthreads()` : the number of CPU threads to use.
+  Defaults to `1` unless [`DataFrames.nthreads!`](@ref) has been called with
+  a different value. Passing a value higher than 1 currently has an effect only
+  for some optimized grouped reductions. Values higher than `Threads.nthreads()`
+  will be replaced with that value.
 
 # Examples
 ```jldoctest

src/groupeddataframe/fastaggregates.jl

@@ -290,7 +290,7 @@ groupreduce(f, op, condf::typeof(!ismissing), adjust, checkempty::Bool,
                  f, op, condf, adjust, checkempty, incol, gd, nthreads)
 
 (r::Reduce)(incol::AbstractVector, gd::GroupedDataFrame;
-            nthreads::Int=1) =
+            nthreads::Int=nthreads()) =
     groupreduce((x, i) -> x, r.op, r.condf, r.adjust, r.checkempty, incol, gd, nthreads)
 
 # this definition is missing in Julia 1.0 LTS and is required by aggregation for var
@@ -300,7 +300,7 @@ if VERSION < v"1.1"
 end
 
 function (agg::Aggregate{typeof(var)})(incol::AbstractVector, gd::GroupedDataFrame;
-                                       nthreads::Int=1)
+                                       nthreads::Int=nthreads())
     means = groupreduce((x, i) -> x, Base.add_sum, agg.condf, /, false,
                         incol, gd, nthreads)
     # !ismissing check is purely an optimization to avoid a copy later
@@ -316,7 +316,7 @@ function (agg::Aggregate{typeof(var)})(incol::AbstractVector, gd::GroupedDataFra
 end
 
 function (agg::Aggregate{typeof(std)})(incol::AbstractVector, gd::GroupedDataFrame;
-                                       nthreads::Int=1)
+                                       nthreads::Int=nthreads())
     outcol = Aggregate(var, agg.condf)(incol, gd; nthreads=nthreads)
     if eltype(outcol) <: Union{Missing, Rational}
         return sqrt.(outcol)
@@ -329,7 +329,7 @@ for f in (:first, :last)
     # Without using @eval the presence of a keyword argument triggers a Julia bug
     @eval begin
         function (agg::Aggregate{typeof($f)})(incol::AbstractVector, gd::GroupedDataFrame;
-                                              nthreads::Int=1)
+                                              nthreads::Int=nthreads())
             n = length(gd)
             outcol = similar(incol, n)
             fillfirst!(agg.condf, outcol, incol, gd, rev=agg.f === last)
@@ -343,7 +343,7 @@ for f in (:first, :last)
 end
 
 function (agg::Aggregate{typeof(length)})(incol::AbstractVector, gd::GroupedDataFrame;
-                                          nthreads::Int=1)
+                                          nthreads::Int=nthreads())
     if getfield(gd, :idx) === nothing
         lens = zeros(Int, length(gd))
         @inbounds for gix in gd.groups

src/groupeddataframe/splitapplycombine.jl

@@ -627,7 +627,7 @@ end
 
 function combine(f::Base.Callable, gd::GroupedDataFrame;
                  keepkeys::Bool=true, ungroup::Bool=true, renamecols::Bool=true,
-                 nthreads::Int=1)
+                 nthreads::Int=nthreads())
     if f isa Colon
         throw(ArgumentError("First argument must be a transformation if the second argument is a GroupedDataFrame"))
     end
@@ -637,22 +637,22 @@ end
 
 combine(f::Pair, gd::GroupedDataFrame;
         keepkeys::Bool=true, ungroup::Bool=true, renamecols::Bool=true,
-        nthreads::Int=1) =
+        nthreads::Int=nthreads()) =
     throw(ArgumentError("First argument must be a transformation if the second argument is a GroupedDataFrame. " *
                         "You can pass a `Pair` as the second argument of the transformation. If you want the return " *
                         "value to be processed as having multiple columns add `=> AsTable` suffix to the pair."))
 
 combine(gd::GroupedDataFrame,
         cs::Union{Pair, Base.Callable, ColumnIndex, MultiColumnIndex}...;
         keepkeys::Bool=true, ungroup::Bool=true, renamecols::Bool=true,
-        nthreads::Int=1) =
+        nthreads::Int=nthreads()) =
     _combine_prepare(gd, cs..., keepkeys=keepkeys, ungroup=ungroup,
                      copycols=true, keeprows=false, renamecols=renamecols,
                      nthreads=nthreads)
 
 function select(f::Base.Callable, gd::GroupedDataFrame; copycols::Bool=true,
                 keepkeys::Bool=true, ungroup::Bool=true, renamecols::Bool=true,
-                nthreads::Int=1)
+                nthreads::Int=nthreads())
     if f isa Colon
         throw(ArgumentError("First argument must be a transformation if the second argument is a grouped data frame"))
     end
@@ -662,14 +662,14 @@ end
 
 
 select(gd::GroupedDataFrame, args...; copycols::Bool=true, keepkeys::Bool=true,
-       ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=1) =
+       ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=nthreads()) =
     _combine_prepare(gd, args..., copycols=copycols, keepkeys=keepkeys,
                      ungroup=ungroup, keeprows=true, renamecols=renamecols,
                      nthreads=nthreads)
 
 function transform(f::Base.Callable, gd::GroupedDataFrame; copycols::Bool=true,
                    keepkeys::Bool=true, ungroup::Bool=true, renamecols::Bool=true,
-                   nthreads::Int=1)
+                   nthreads::Int=nthreads())
     if f isa Colon
         throw(ArgumentError("First argument must be a transformation if the second argument is a grouped data frame"))
     end
@@ -679,7 +679,7 @@ end
 
 function transform(gd::GroupedDataFrame, args...; copycols::Bool=true,
                    keepkeys::Bool=true, ungroup::Bool=true, renamecols::Bool=true,
-                   nthreads::Int=1)
+                   nthreads::Int=nthreads())
     res = select(gd, :, args..., copycols=copycols, keepkeys=keepkeys,
                  ungroup=ungroup, renamecols=renamecols, nthreads=nthreads)
     # res can be a GroupedDataFrame based on DataFrame or a DataFrame,
@@ -689,31 +689,31 @@ function transform(gd::GroupedDataFrame, args...; copycols::Bool=true,
 end
 
 function select!(f::Base.Callable, gd::GroupedDataFrame;
-                 ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=1)
+                 ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=nthreads())
     if f isa Colon
         throw(ArgumentError("First argument must be a transformation if the second argument is a grouped data frame"))
     end
     return select!(gd, f, ungroup=ungroup, nthreads=nthreads)
 end
 
 function select!(gd::GroupedDataFrame{DataFrame}, args...;
-                 ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=1)
+                 ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=nthreads())
     newdf = select(gd, args..., copycols=false, renamecols=renamecols, nthreads=nthreads)
     df = parent(gd)
     _replace_columns!(df, newdf)
     return ungroup ? df : gd
 end
 
 function transform!(f::Base.Callable, gd::GroupedDataFrame;
-                    ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=1)
+                    ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=nthreads())
     if f isa Colon
         throw(ArgumentError("First argument must be a transformation if the second argument is a grouped data frame"))
     end
     return transform!(gd, f, ungroup=ungroup, nthreads=nthreads)
 end
 
 function transform!(gd::GroupedDataFrame{DataFrame}, args...;
-                    ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=1)
+                    ungroup::Bool=true, renamecols::Bool=true, nthreads::Int=nthreads())
     newdf = select(gd, :, args..., copycols=false, renamecols=renamecols, nthreads=nthreads)
     df = parent(gd)
     select!(newdf, propertynames(df), :)

src/other/utils.jl

@@ -82,4 +82,29 @@ else
     using Compat: ComposedFunction
 end
 
-funname(c::ComposedFunction) = Symbol(funname(c.outer), :_, funname(c.inner))
+funname(c::ComposedFunction) = Symbol(funname(c.outer), :_, funname(c.inner))
+
+const NTHREADS = Ref(1)
+
+"""
+    DataFrames.nthreads()
+
+Return the default value for the `nthreads` argument, which determines the number
+of CPU threads used by functions that support it when not specified explicitly.
+
+Defaults to 1. Call [`DataFrames.nthreads!`](@ref) to adjust the value.
+"""
+nthreads() = DataFrames.NTHREADS[]
+
+"""
+    DataFrames.nthreads!(n::Int)
+
+Set to `n` the default value for the `nthreads` argument, which determines the number
+of CPU threads used by functions that support it when not specified explicitly.
+
+Use [`DataFrames.nthreads`](@ref) to access the value.
+"""
+function nthreads!(n::Int)
+    n > 0 || throw(ArgumentError("n must be equal to or greater than 1 (got $n)"))
+    DataFrames.NTHREADS[] = n
+end

test/utils.jl

@@ -94,4 +94,14 @@ end
           :sum_skipmissing_div12
 end
 
+@test "nthreads and nthreads!" begin
+    @test DataFrames.nthreads() == 1
+    @test DataFrames.nthreads!(2) == 2
+    @test DataFrames.nthreads() == 2
+    @test DataFrames.nthreads!(1) == 1 # reset to default
+    @test_throws ArgumentError DataFrames.nthreads!(0)
+    @test DataFrames.nthreads() == 1
+end
+end
+
 end # module