Skip to content

[enhancement] adding csr online support to oneDAL basic_statistics #3715

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
icfaust wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

[enhancement] adding csr online support to oneDAL basic_statistics #3715

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b654cc1
adding csr online support
icfaust Jul 24, 2026
09e0c93
forgotten additions:
icfaust Jul 24, 2026
ab0a0ce
move based on suggestions
icfaust Jul 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion cpp/daal/src/algorithms/low_order_moments/low_order_moments_csr_fast_online_fpt_cpu.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ namespace internal
{
template class OnlineContainer<DAAL_FPTYPE, fastCSR, DAAL_CPU>;

template class LowOrderMomentsOnlineKernel<DAAL_FPTYPE, fastCSR, DAAL_CPU>;
template class DAAL_EXPORT LowOrderMomentsOnlineKernel<DAAL_FPTYPE, fastCSR, DAAL_CPU>;
} // namespace internal
} // namespace low_order_moments
} // namespace algorithms
Expand Down
64 changes: 40 additions & 24 deletions cpp/oneapi/dal/algo/basic_statistics/backend/cpu/finalize_compute_kernel_dense.cpp
View file
Open in desktop

@icfaust icfaust Jul 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would need to rename this file based on its use for csr support as well (following the precedent of basic_statistics for batch jobs).

Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
namespace oneapi::dal::basic_statistics::backend {

using dal::backend::context_cpu;
using method_t = method::dense;
using task_t = task::compute;
using input_t = compute_input<task_t>;
using result_t = compute_result<task_t>;
Expand All @@ -41,11 +40,23 @@ namespace daal_lom = daal::algorithms::low_order_moments;
namespace interop = dal::backend::interop;
namespace bk = dal::backend;

template <typename Float, daal::internal::CpuType Cpu>
template <daal_lom::Method Value>
using daal_method_constant = std::integral_constant<daal_lom::Method, Value>;

@icfaust icfaust Jul 24, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This follows on precedent in kmeans.


template <typename Method>
struct to_daal_method;

template <>
struct to_daal_method<method::dense> : daal_method_constant<daal_lom::defaultDense> {};

template <>
struct to_daal_method<method::sparse> : daal_method_constant<daal_lom::fastCSR> {};

template <typename Float, daal::internal::CpuType Cpu, typename Method>
using daal_lom_online_kernel_t =
daal_lom::internal::LowOrderMomentsOnlineKernel<Float, daal_lom::defaultDense, Cpu>;
daal_lom::internal::LowOrderMomentsOnlineKernel<Float, to_daal_method<Method>::value, Cpu>;

template <typename Float, typename Task>
template <typename Float, typename Method, typename Task>
static compute_result<Task> call_daal_kernel_finalize_compute(
const context_cpu& ctx,
const descriptor_t& desc,
Expand Down Expand Up @@ -80,19 +91,22 @@ static compute_result<Task> call_daal_kernel_finalize_compute(
auto daal_stdev = interop::allocate_daal_homogen_table<Float>(1, column_count);
auto daal_variation = interop::allocate_daal_homogen_table<Float>(1, column_count);
if (result_ids == daal_lom::estimatesMeanVariance || result_ids == daal_lom::estimatesAll) {
interop::status_to_exception(
interop::call_daal_kernel_finalize_compute<Float, daal_lom_online_kernel_t>(
ctx,
daal_partial_obs.get(),
daal_partial_sums.get(),
daal_partial_sum_squares.get(),
daal_partial_sum_squares_centered.get(),
daal_means.get(),
daal_rawt.get(),
daal_variance.get(),
daal_stdev.get(),
daal_variation.get(),
&daal_parameter));
interop::status_to_exception(dal::backend::dispatch_by_cpu(ctx, [&](auto cpu) {
return daal_lom_online_kernel_t<
Float,
oneapi::dal::backend::interop::to_daal_cpu_type<decltype(cpu)>::value,
Method>()
.finalizeCompute(daal_partial_obs.get(),

@icfaust icfaust Jul 24, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

required due to limitations in call_daal_kernel_finalize_compute to the default method (I didn't understand this at first, but have verified).

daal_partial_sums.get(),
daal_partial_sum_squares.get(),
daal_partial_sum_squares_centered.get(),
daal_means.get(),
daal_rawt.get(),
daal_variance.get(),
daal_stdev.get(),
daal_variation.get(),
&daal_parameter);
}));
}
compute_result<Task> res;
res.set_result_options(desc.get_result_options());
Expand Down Expand Up @@ -132,24 +146,26 @@ static compute_result<Task> call_daal_kernel_finalize_compute(
return res;
}

template <typename Float, typename Task>
template <typename Float, typename Method, typename Task>
static compute_result<Task> finalize_compute(const context_cpu& ctx,
const descriptor_t& desc,
const partial_compute_result<Task>& input) {
return call_daal_kernel_finalize_compute<Float, Task>(ctx, desc, input);
return call_daal_kernel_finalize_compute<Float, Method, Task>(ctx, desc, input);
}

template <typename Float>
struct finalize_compute_kernel_cpu<Float, method_t, task_t> {
template <typename Float, typename Method>
struct finalize_compute_kernel_cpu<Float, Method, task_t> {
compute_result<task::compute> operator()(
const context_cpu& ctx,
const descriptor_t& desc,
const partial_compute_result<task::compute>& input) const {
return finalize_compute<Float, task::compute>(ctx, desc, input);
return finalize_compute<Float, Method, task::compute>(ctx, desc, input);
}
};

template struct finalize_compute_kernel_cpu<float, method_t, task_t>;
template struct finalize_compute_kernel_cpu<double, method_t, task_t>;
template struct finalize_compute_kernel_cpu<float, method::dense, task_t>;
template struct finalize_compute_kernel_cpu<double, method::dense, task_t>;
template struct finalize_compute_kernel_cpu<float, method::sparse, task_t>;
template struct finalize_compute_kernel_cpu<double, method::sparse, task_t>;

} // namespace oneapi::dal::basic_statistics::backend
85 changes: 52 additions & 33 deletions cpp/oneapi/dal/algo/basic_statistics/backend/cpu/partial_compute_kernel_dense.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,6 @@
namespace oneapi::dal::basic_statistics::backend {

using dal::backend::context_cpu;
using method_t = method::dense;
using task_t = task::compute;
using input_t = partial_compute_input<task_t>;
using result_t = partial_compute_result<task_t>;
Expand All @@ -41,8 +40,24 @@ using descriptor_t = detail::descriptor_base<task_t>;
namespace daal_lom = daal::algorithms::low_order_moments;
namespace interop = dal::backend::interop;

template <typename Float, daal::internal::CpuType Cpu>
template <daal_lom::Method Value>
using daal_method_constant = std::integral_constant<daal_lom::Method, Value>;

template <typename Method>
struct to_daal_method;

template <>
struct to_daal_method<method::dense> : daal_method_constant<daal_lom::defaultDense> {};

template <>
struct to_daal_method<method::sparse> : daal_method_constant<daal_lom::fastCSR> {};

template <typename Float, daal::internal::CpuType Cpu, typename Method>
using daal_lom_online_kernel_t =
daal_lom::internal::LowOrderMomentsOnlineKernel<Float, to_daal_method<Method>::value, Cpu>;

template <typename Float, daal::internal::CpuType Cpu>
using daal_lom_online_dense_kernel_t =
daal_lom::internal::LowOrderMomentsOnlineKernel<Float, daal_lom::defaultDense, Cpu>;

template <typename Float, typename Task>
Expand Down Expand Up @@ -146,11 +161,11 @@ result_t call_daal_kernel_with_weights(const context_cpu& ctx,
}
{
interop::status_to_exception(
interop::call_daal_kernel<Float, daal_lom_online_kernel_t>(ctx,
daal_data.get(),
&daal_partial,
&daal_parameter,
is_online));
interop::call_daal_kernel<Float, daal_lom_online_dense_kernel_t>(ctx,
daal_data.get(),
&daal_partial,
&daal_parameter,
is_online));
}
auto result = get_partial_result<Float, task_t>(daal_partial, desc);

Expand All @@ -159,18 +174,18 @@ result_t call_daal_kernel_with_weights(const context_cpu& ctx,
else {
{
interop::status_to_exception(
interop::call_daal_kernel<Float, daal_lom_online_kernel_t>(ctx,
daal_data.get(),
&daal_partial,
&daal_parameter,
is_online));
interop::call_daal_kernel<Float, daal_lom_online_dense_kernel_t>(ctx,
daal_data.get(),
&daal_partial,
&daal_parameter,
is_online));
}
auto result = get_partial_result<Float, task_t>(daal_partial, desc);
return result;
}
}

template <typename Float, typename Task>
template <typename Float, typename Method, typename Task>
result_t call_daal_kernel_without_weights(const context_cpu& ctx,
const descriptor_t& desc,
const partial_compute_input<Task>& input) {
Expand Down Expand Up @@ -223,52 +238,56 @@ result_t call_daal_kernel_without_weights(const context_cpu& ctx,
daal_partial.set(daal_lom::PartialResultId::partialSumSquares,
daal_partial_sum_squares);
}
interop::status_to_exception(
interop::call_daal_kernel<Float, daal_lom_online_kernel_t>(ctx,
daal_data.get(),
&daal_partial,
&daal_parameter,
is_online));
interop::status_to_exception(dal::backend::dispatch_by_cpu(ctx, [&](auto cpu) {
return daal_lom_online_kernel_t<
Float,
oneapi::dal::backend::interop::to_daal_cpu_type<decltype(cpu)>::value,
Method>()
.compute(daal_data.get(), &daal_partial, &daal_parameter, is_online);
}));
auto result = get_partial_result<Float, task_t>(daal_partial, desc);
return result;
}
else {
{
interop::status_to_exception(
interop::call_daal_kernel<Float, daal_lom_online_kernel_t>(ctx,
daal_data.get(),
&daal_partial,
&daal_parameter,
is_online));
interop::status_to_exception(dal::backend::dispatch_by_cpu(ctx, [&](auto cpu) {
return daal_lom_online_kernel_t<
Float,
oneapi::dal::backend::interop::to_daal_cpu_type<decltype(cpu)>::value,
Method>()
.compute(daal_data.get(), &daal_partial, &daal_parameter, is_online);
}));
}
auto result = get_partial_result<Float, task_t>(daal_partial, desc);
return result;
}
}

template <typename Float, typename Task>
template <typename Float, typename Method, typename Task>
static partial_compute_result<Task> partial_compute(const context_cpu& ctx,
const descriptor_t& desc,
const partial_compute_input<Task>& input) {
if (input.get_weights().has_data()) {
return call_daal_kernel_with_weights<Float>(ctx, desc, input);
return call_daal_kernel_with_weights<Float, Task>(ctx, desc, input);
}
else {
return call_daal_kernel_without_weights<Float, Task>(ctx, desc, input);
return call_daal_kernel_without_weights<Float, Method, Task>(ctx, desc, input);
}
}

template <typename Float>
struct partial_compute_kernel_cpu<Float, method_t, task_t> {
template <typename Float, typename Method>
struct partial_compute_kernel_cpu<Float, Method, task_t> {
partial_compute_result<task::compute> operator()(
const context_cpu& ctx,
const descriptor_t& desc,
const partial_compute_input<task::compute>& input) const {
return partial_compute<Float, task::compute>(ctx, desc, input);
return partial_compute<Float, Method, task::compute>(ctx, desc, input);
}
};

template struct partial_compute_kernel_cpu<float, method_t, task_t>;
template struct partial_compute_kernel_cpu<double, method_t, task_t>;
template struct partial_compute_kernel_cpu<float, method::dense, task_t>;
template struct partial_compute_kernel_cpu<double, method::dense, task_t>;
template struct partial_compute_kernel_cpu<float, method::sparse, task_t>;
template struct partial_compute_kernel_cpu<double, method::sparse, task_t>;

} // namespace oneapi::dal::basic_statistics::backend
2 changes: 2 additions & 0 deletions cpp/oneapi/dal/algo/basic_statistics/detail/finalize_compute_ops.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ struct finalize_compute_ops_dispatcher<Policy, Float, Method, Task> {

INSTANTIATE(float, method::dense, task::compute)
INSTANTIATE(double, method::dense, task::compute)
INSTANTIATE(float, method::sparse, task::compute)
INSTANTIATE(double, method::sparse, task::compute)

} // namespace v1
} // namespace oneapi::dal::basic_statistics::detail
2 changes: 2 additions & 0 deletions cpp/oneapi/dal/algo/basic_statistics/detail/finalize_compute_ops_dpc.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,8 @@ struct finalize_compute_ops_dispatcher<Policy, Float, Method, Task> {

INSTANTIATE(float, method::dense, task::compute)
INSTANTIATE(double, method::dense, task::compute)
INSTANTIATE(float, method::sparse, task::compute)
INSTANTIATE(double, method::sparse, task::compute)

} // namespace v1
} // namespace oneapi::dal::basic_statistics::detail
2 changes: 2 additions & 0 deletions cpp/oneapi/dal/algo/basic_statistics/detail/partial_compute_ops.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,8 @@ struct partial_compute_ops_dispatcher<Policy, Float, Method, Task> {

INSTANTIATE(float, method::dense, task::compute)
INSTANTIATE(double, method::dense, task::compute)
INSTANTIATE(float, method::sparse, task::compute)
INSTANTIATE(double, method::sparse, task::compute)

} // namespace v1
} // namespace oneapi::dal::basic_statistics::detail
2 changes: 2 additions & 0 deletions cpp/oneapi/dal/algo/basic_statistics/detail/partial_compute_ops_dpc.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ struct partial_compute_ops_dispatcher<Policy, Float, Method, Task> {

INSTANTIATE(float, method::dense, task::compute)
INSTANTIATE(double, method::dense, task::compute)
INSTANTIATE(float, method::sparse, task::compute)
INSTANTIATE(double, method::sparse, task::compute)

} // namespace v1
} // namespace oneapi::dal::basic_statistics::detail
50 changes: 50 additions & 0 deletions cpp/oneapi/dal/algo/basic_statistics/test/fixture.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,56 @@ class basic_statistics_test : public te::crtp_algo_fixture<TestType, Derived> {
return (data.row_count_ > 100 || data.column_count_ > 100) && policy.is_cpu();
}

std::vector<csr_table> split_csr_by_rows(const csr_table& table, std::int64_t split_count) {

@icfaust icfaust Jul 24, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I need help here. I am not well versed in catch2 (only basic knowledge).

@Vika-F Vika-F Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What kind of help?
This implementation looks reasonable from the first glance for CPU-only usage.

@Vika-F Vika-F Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please move it to the common place: https://github.com/uxlfoundation/oneDAL/blob/main/cpp/oneapi/dal/test/engine/tables.hpp

ONEDAL_ASSERT(split_count > 0);
const std::int64_t row_count = table.get_row_count();
const std::int64_t column_count = table.get_column_count();
const std::int64_t block_size_regular = row_count / split_count;
const std::int64_t block_size_tail = row_count % split_count;
const auto indexing = table.get_indexing();

csr_accessor<const float_t> accessor(table);
std::vector<csr_table> result(split_count);

std::int64_t row_offset = 0;
for (std::int64_t i = 0; i < split_count; ++i) {
const std::int64_t tail = std::int64_t(i + 1 == split_count) * block_size_tail;
const std::int64_t block_size = block_size_regular + tail;
if (block_size <= 0) {
result[i] = csr_table{};
row_offset += block_size;
continue;
}
const auto [data_arr, col_arr, row_arr] =
accessor.pull({ row_offset, row_offset + block_size }, indexing);
result[i] =
csr_table::wrap(data_arr, col_arr, row_arr, column_count, indexing);
row_offset += block_size;
}
return result;
}

void csr_online_general_checks(const te::csr_table_builder<>& data,
bs::result_option_id compute_mode,
std::int64_t nBlocks) {
CAPTURE(compute_mode, nBlocks);
const auto bs_desc =
bs::descriptor<float_t, basic_statistics::method::sparse>{}.set_result_options(
compute_mode);
const auto csr_full = data.build_csr_table(this->get_policy());
const auto dense_full = data.build_dense_table(this->get_policy());

const auto blocks = split_csr_by_rows(csr_full, nBlocks);
dal::basic_statistics::partial_compute_result<> partial_result;
for (std::int64_t i = 0; i < nBlocks; ++i) {
partial_result = this->partial_compute(bs_desc, partial_result, blocks[i]);
}
auto compute_result = this->finalize_compute(bs_desc, partial_result);
table weights;
check_compute_result(compute_mode, dense_full, weights, compute_result);
check_for_exception_for_non_requested_results(compute_mode, compute_result);
}

void online_general_checks(const te::dataframe& data_fr,
std::shared_ptr<te::dataframe> weights_fr,
bs::result_option_id compute_mode,
Expand Down
25 changes: 25 additions & 0 deletions cpp/oneapi/dal/algo/basic_statistics/test/online.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,4 +62,29 @@ TEMPLATE_LIST_TEST_M(basic_statistics_online_test,
this->online_general_checks(data, weights, compute_mode, nBlocks);
}

TEMPLATE_LIST_TEST_M(basic_statistics_online_test,
"basic_statistics common CSR online flow",
"[basic_statistics][integration][online]",
basic_statistics_sparse_types) {
SKIP_IF(this->get_policy().is_gpu());
SKIP_IF(this->not_float64_friendly());
const float nnz_fraction = 0.05;
this->data_indexing_ = GENERATE(sparse_indexing::zero_based, sparse_indexing::one_based);
const auto data =
GENERATE_COPY(te::csr_table_builder(5, 5, nnz_fraction, this->data_indexing_),
te::csr_table_builder(7, 10, nnz_fraction, this->data_indexing_),
te::csr_table_builder(100, 100, nnz_fraction, this->data_indexing_));
SKIP_IF(this->not_cpu_friendly(data));

const std::int64_t nBlocks = GENERATE(1, 3, 5);

const bs::result_option_id res_min_max = result_options::min | result_options::max;
const bs::result_option_id res_mean_varc = result_options::mean | result_options::variance;
const bs::result_option_id res_all =
bs::result_option_id(dal::result_option_id_base(mask_full));
const bs::result_option_id compute_mode = GENERATE_COPY(res_min_max, res_mean_varc, res_all);

this->csr_online_general_checks(data, compute_mode, nBlocks);
}

} // namespace oneapi::dal::basic_statistics::test
Loading