diff --git a/Framework/AnalysisSupport/CMakeLists.txt b/Framework/AnalysisSupport/CMakeLists.txt
index 956c4a44c5684..f69e7ebc2110c 100644
--- a/Framework/AnalysisSupport/CMakeLists.txt
+++ b/Framework/AnalysisSupport/CMakeLists.txt
@@ -19,6 +19,7 @@ endif()
 o2_add_library(FrameworkOnDemandTablesSupport
                SOURCES src/OnDemandPlugin.cxx
                        src/AODReaderHelpers.cxx
+                       src/AODSliceHelpers.cxx
                PRIVATE_INCLUDE_DIRECTORIES ${CMAKE_CURRENT_LIST_DIR}/src
                PUBLIC_LINK_LIBRARIES O2::Framework ${EXTRA_TARGETS})
 
diff --git a/Framework/AnalysisSupport/src/AODSliceHelpers.cxx b/Framework/AnalysisSupport/src/AODSliceHelpers.cxx
new file mode 100644
index 0000000000000..deee30343762d
--- /dev/null
+++ b/Framework/AnalysisSupport/src/AODSliceHelpers.cxx
@@ -0,0 +1,66 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#include "AODSliceHelpers.h"
+
+#include "Framework/ArrowTableSlicingCache.h"
+#include "Framework/ConfigParamRegistry.h"
+#include "Framework/DanglingEdgesContext.h"
+#include "Framework/DataAllocator.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/InputRecord.h"
+#include "Framework/TableConsumer.h"
+
+namespace o2::framework::helpers
+{
+namespace
+{
+Entry sourceEntry(InputSpec const& spec)
+{
+  auto source = DataSpecUtils::fromMetadataString(std::ranges::find_if(spec.metadata, [](ConfigParamSpec const& cps) { return cps.name.starts_with("slice-source"); })->defaultValue.get<std::string>());
+  return {source.binding, DataSpecUtils::asConcreteDataMatcher(source), std::ranges::find_if(spec.metadata, [](ConfigParamSpec const& cps) { return cps.name.starts_with("slice-key"); })->defaultValue.get<std::string>()};
+}
+
+struct Sliceable {
+  Entry entry;
+  ConcreteDataMatcher output;
+  bool sorted;
+
+  explicit Sliceable(InputSpec const& spec)
+    : entry{sourceEntry(spec)},
+      output{DataSpecUtils::asConcreteDataMatcher(spec)},
+      sorted{std::ranges::find_if(spec.metadata, [](ConfigParamSpec const& cps) { return cps.name.starts_with("sorted"); })->defaultValue.get<bool>()}
+  {
+  }
+
+  std::shared_ptr<arrow::Table> materialize(ProcessingContext& pc) const
+  {
+    auto source = pc.inputs().get<TableConsumer>(entry.matcher)->asArrowTable();
+    return sorted ? SliceInfo::makeSorted(entry, source) : SliceInfo::makeUnsorted(entry, source);
+  }
+};
+} // namespace
+
+AlgorithmSpec AODSliceHelpers::arrowTablesSlicerCallback(ConfigContext const& /*ctx*/)
+{
+  return AlgorithmSpec::InitCallback{[](InitContext& ic) {
+    // each slicer handles the group of slice infos for the tables from a single provider
+    auto const& requested = ic.services().get<DanglingEdgesContext>().slicerGroups[ic.options().get<int>("slicer-group")];
+    std::vector<Sliceable> sliceables;
+    sliceables.reserve(requested.size());
+    std::ranges::transform(requested, std::back_inserter(sliceables), [](auto const& i) { return Sliceable{i}; });
+    return [sliceables](ProcessingContext& pc) {
+      auto outputs = pc.outputs();
+      std::ranges::for_each(sliceables, [&pc, &outputs](auto const& sliceable) { outputs.adopt(Output{sliceable.output.origin, sliceable.output.description, sliceable.output.subSpec}, sliceable.materialize(pc)); });
+    };
+  }};
+}
+} // namespace o2::framework::helpers
diff --git a/Framework/AnalysisSupport/src/AODSliceHelpers.h b/Framework/AnalysisSupport/src/AODSliceHelpers.h
new file mode 100644
index 0000000000000..59d0448cf9f8b
--- /dev/null
+++ b/Framework/AnalysisSupport/src/AODSliceHelpers.h
@@ -0,0 +1,25 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#ifndef AODSLICEHELPERS_H
+#define AODSLICEHELPERS_H
+
+#include "Framework/AlgorithmSpec.h"
+namespace o2::framework::helpers
+{
+
+struct AODSliceHelpers {
+  static AlgorithmSpec arrowTablesSlicerCallback(ConfigContext const& /*ctx*/);
+};
+
+} // namespace o2::framework::helpers
+
+#endif // AODSLICEHELPERS_H
diff --git a/Framework/AnalysisSupport/src/OnDemandPlugin.cxx b/Framework/AnalysisSupport/src/OnDemandPlugin.cxx
index 9438f9bf69c96..19bd7261ecc99 100644
--- a/Framework/AnalysisSupport/src/OnDemandPlugin.cxx
+++ b/Framework/AnalysisSupport/src/OnDemandPlugin.cxx
@@ -11,6 +11,7 @@
 #include "Framework/Plugins.h"
 #include "Framework/AlgorithmSpec.h"
 #include "AODReaderHelpers.h"
+#include "AODSliceHelpers.h"
 
 struct ExtendedTableSpawner : o2::framework::AlgorithmPlugin {
   o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& config) override
@@ -26,7 +27,15 @@ struct IndexTableBuilder : o2::framework::AlgorithmPlugin {
   }
 };
 
+struct ArrowTableSlicer : o2::framework::AlgorithmPlugin {
+  o2::framework::AlgorithmSpec create(o2::framework::ConfigContext const& config) override
+  {
+    return o2::framework::helpers::AODSliceHelpers::arrowTablesSlicerCallback(config);
+  }
+};
+
 DEFINE_DPL_PLUGINS_BEGIN
 DEFINE_DPL_PLUGIN_INSTANCE(ExtendedTableSpawner, CustomAlgorithm);
 DEFINE_DPL_PLUGIN_INSTANCE(IndexTableBuilder, CustomAlgorithm);
+DEFINE_DPL_PLUGIN_INSTANCE(ArrowTableSlicer, CustomAlgorithm);
 DEFINE_DPL_PLUGINS_END
diff --git a/Framework/Core/include/Framework/ASoA.h b/Framework/Core/include/Framework/ASoA.h
index 4f1ef6bdeda57..bb6c1c20ae78e 100644
--- a/Framework/Core/include/Framework/ASoA.h
+++ b/Framework/Core/include/Framework/ASoA.h
@@ -15,8 +15,8 @@
 #if defined(__CLING__)
 #error "Please do not include this file in ROOT dictionary generation"
 #endif
+#include "Framework/InputSpec.h"
 #include "Framework/Concepts.h"
-#include "Framework/ConcreteDataMatcher.h"
 #include "Framework/Pack.h"                   // IWYU pragma: export
 #include "Framework/FunctionalHelpers.h"      // IWYU pragma: export
 #include "Headers/DataHeader.h"               // IWYU pragma: export
@@ -1351,7 +1351,7 @@ static constexpr std::pair<bool, framework::ConcreteDataMatcher> hasKeyM(std::st
 }
 
 void notFoundColumn(const char* label, const char* key);
-void missingOptionalPreslice(const char* label, const char* key);
+void missingPreslice(const char* label, const char* key);
 
 template <with_originals T, bool OPT = false>
 static constexpr std::string getLabelFromTypeForKey(std::string_view key)
@@ -1421,7 +1421,6 @@ namespace o2::framework
 /// tracks origin in bindingKey matcher to handle the correct arguments
 struct PreslicePolicyBase {
   static constexpr void isPreslicePolicy() {};
-  const std::string binding;
   Entry bindingKey;
 
   bool isMissing() const;
@@ -1448,29 +1447,27 @@ struct PresliceBase : public Policy {
   constexpr static bool optional = OPT;
   using target_t = T;
   using policy_t = Policy;
-  const std::string binding;
 
   PresliceBase(expressions::BindingNode index_)
-    : Policy{PreslicePolicyBase{{o2::soa::getLabelFromTypeForKey<T, OPT>(std::string{index_.name})}, Entry(o2::soa::getLabelFromTypeForKey<T, OPT>(std::string{index_.name}), o2::soa::getMatcherFromTypeForKey<T, OPT>(std::string{index_.name}), std::string{index_.name})}, {}}
+    : Policy{Entry(
+        o2::soa::getLabelFromTypeForKey<T, true>(std::string{index_.name}),
+        o2::soa::getMatcherFromTypeForKey<T, true>(std::string{index_.name}),
+        std::string{index_.name})}
   {
   }
 
   o2::soa::ArrowTableRef getSliceFor(int value, o2::soa::ArrowTableRef const& input) const
   {
-    if constexpr (OPT) {
-      if (Policy::isMissing()) {
-        return {nullptr, {0, 0}};
-      }
+    if (Policy::isMissing()) {
+      return {nullptr, {0, 0}};
     }
     return Policy::getSliceFor(value, input);
   }
 
   std::span<const int64_t> getSliceFor(int value) const
   {
-    if constexpr (OPT) {
-      if (Policy::isMissing()) {
-        return {};
-      }
+    if (Policy::isMissing()) {
+      return {};
     }
     return Policy::getSliceFor(value);
   }
@@ -1526,10 +1523,8 @@ template <typename T, typename C, typename Policy, bool OPT>
   requires std::same_as<Policy, framework::PreslicePolicySorted> && (o2::soa::is_binding_compatible_v<C, T>())
 auto doSliceBy(T const* table, o2::framework::PresliceBase<C, Policy, OPT> const& container, int value)
 {
-  if constexpr (OPT) {
-    if (container.isMissing()) {
-      missingOptionalPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.key.c_str());
-    }
+  if (container.isMissing()) {
+    missingPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.key.c_str());
   }
   auto out = container.getSliceFor(value, table->asArrowTableRef());
   auto t = typename T::self_t({out});
@@ -1568,10 +1563,8 @@ template <typename T, typename C, typename Policy, bool OPT>
   requires std::same_as<Policy, framework::PreslicePolicyGeneral> && (o2::soa::is_binding_compatible_v<C, T>())
 auto doSliceBy(T const* table, o2::framework::PresliceBase<C, Policy, OPT> const& container, int value)
 {
-  if constexpr (OPT) {
-    if (container.isMissing()) {
-      missingOptionalPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.key.c_str());
-    }
+  if (container.isMissing()) {
+    missingPreslice(getLabelFromType<std::decay_t<T>>().data(), container.bindingKey.key.c_str());
   }
   auto selection = container.getSliceFor(value);
   return doSliceByHelper(table, selection);
@@ -1601,10 +1594,8 @@ template <soa::is_filtered_table T, typename C, bool OPT>
   requires(o2::soa::is_binding_compatible_v<C, T>())
 auto doFilteredSliceBy(T const* table, o2::framework::PresliceBase<C, framework::PreslicePolicySorted, OPT> const& container, int value)
 {
-  if constexpr (OPT) {
-    if (container.isMissing()) {
-      missingOptionalPreslice(getLabelFromType<T>().data(), container.bindingKey.key.c_str());
-    }
+  if (container.isMissing()) {
+    missingPreslice(getLabelFromType<T>().data(), container.bindingKey.key.c_str());
   }
   auto slice = container.getSliceFor(value, table->asArrowTableRef());
   return prepareFilteredSlice(table, slice);
diff --git a/Framework/Core/include/Framework/AnalysisHelpers.h b/Framework/Core/include/Framework/AnalysisHelpers.h
index 6071e8291e387..f0d10cf6d8576 100644
--- a/Framework/Core/include/Framework/AnalysisHelpers.h
+++ b/Framework/Core/include/Framework/AnalysisHelpers.h
@@ -37,6 +37,8 @@ struct InputInfo {
   uint32_t hash;
   std::vector<std::pair<int, ConcreteDataMatcher>> matchers;
 };
+
+void updateInputInfos(std::vector<InputInfo>& iInfos, ConcreteDataMatcher&& matcher, uint32_t hash, int ai);
 } // namespace o2::framework
 
 namespace o2::soa
diff --git a/Framework/Core/include/Framework/AnalysisManagers.h b/Framework/Core/include/Framework/AnalysisManagers.h
index 8b426576ce556..2e116b53787d2 100644
--- a/Framework/Core/include/Framework/AnalysisManagers.h
+++ b/Framework/Core/include/Framework/AnalysisManagers.h
@@ -627,6 +627,73 @@ bool replaceOrigin(T& presliceGroup, header::DataOrigin const& newOrigin)
   return true;
 }
 
+template <typename T>
+  requires(!is_preslice<T> && !is_preslice_group<T>)
+bool addSlicingInputs(T&, std::vector<InputSpec>&, header::DataOrigin const&)
+{
+  return false;
+}
+
+/// check if any of the tables the sliced type is based on is an input of the task
+template <soa::is_table T>
+bool isSlicedTableInput(std::vector<InputSpec> const& inputs, header::DataOrigin const& newOrigin)
+{
+  auto isInput = [&inputs, &newOrigin](ConcreteDataMatcher matcher) {
+    if ((matcher.origin == header::DataOrigin{"AOD"}) && (newOrigin != header::DataOrigin{"AOD"})) {
+      matcher = replaceOrigin(matcher, newOrigin);
+    }
+    return std::ranges::any_of(inputs, [&matcher](InputSpec const& input) { return DataSpecUtils::match(input, matcher); });
+  };
+  return [&isInput]<size_t... Is>(std::index_sequence<Is...>) {
+    return (isInput(o2::aod::matcher<T::originals[Is]>()) || ...);
+  }(std::make_index_sequence<T::originals.size()>{});
+}
+
+/// all the process function inputs are already added at this point, so a Preslice can only
+/// amend them. Depending on whether the sliced table is an input of the task and whether it has
+/// the index column, there are 4 cases:
+/// 1. no table, no column - likely an incorrect declaration, warning for both Preslice and PresliceOptional
+/// 2. no table, column    - Preslice that never works, or a common declaration in a templated task that is
+///                          not effective in this specialization, warning for Preslice only
+/// 3. table, no column    - the intended case for PresliceOptional, a mistake for Preslice, warning for Preslice only
+/// 4. table, column       - slicing input is added
+template <is_preslice T>
+bool addSlicingInputs(T& preslice, std::vector<InputSpec>& inputs, header::DataOrigin const& newOrigin)
+{
+  using target_t = typename T::target_t;
+  auto const& [binding, matcher, key, enabled] = preslice.bindingKey;
+  if (preslice.isMissing()) {
+    if (!isSlicedTableInput<target_t>(inputs, newOrigin)) {
+      LOGP(warn, "Preslice declared on {} is skipped: {} is not an input of any process function and does not have column {}, the declaration is likely incorrect",
+           o2::soa::getLabelFromType<target_t>(), o2::soa::getLabelFromType<target_t>(), key);
+    } else if constexpr (!T::optional) {
+      LOGP(warn, "Preslice declared on {} is skipped: it does not have column {}, use PresliceOptional if the column is not always expected",
+           o2::soa::getLabelFromType<target_t>(), key);
+    }
+    return true;
+  }
+  if (std::ranges::none_of(inputs, [&matcher](InputSpec const& input) { return DataSpecUtils::match(input, matcher); })) {
+    if constexpr (!T::optional) {
+      LOGP(warn, "Preslice declared on {}/{} ({}) is skipped: {} is not an input of any process function, use PresliceOptional if the declaration is not effective in every specialization of a templated task",
+           binding, key, DataSpecUtils::describe(matcher), binding);
+    }
+    return true;
+  }
+  DataSpecUtils::updateInputList(inputs, inputForEntry(preslice.bindingKey, std::same_as<typename T::policy_t, framework::PreslicePolicySorted>));
+  return true;
+}
+
+template <is_preslice_group T>
+bool addSlicingInputs(T&& presliceGroup, std::vector<InputSpec>& inputs, header::DataOrigin const& newOrigin)
+{
+  homogeneous_apply_refs<true>(
+    [&inputs, &newOrigin](auto& preslice) {
+      return addSlicingInputs(preslice, inputs, newOrigin);
+    },
+    presliceGroup);
+  return true;
+}
+
 template <typename T>
   requires(!is_preslice<T> && !is_preslice_group<T>)
 bool registerCache(T&, Cache&, Cache&)
@@ -638,10 +705,8 @@ template <is_preslice T>
   requires std::same_as<typename T::policy_t, framework::PreslicePolicySorted>
 bool registerCache(T& preslice, Cache& bsks, Cache&)
 {
-  if constexpr (T::optional) {
-    if (preslice.binding == "[MISSING]") {
-      return true;
-    }
+  if (preslice.isMissing()) {
+    return true;
   }
   auto locate = std::find(bsks.begin(), bsks.end(), preslice.getBindingKey());
   if (locate == bsks.end()) {
@@ -656,10 +721,8 @@ template <is_preslice T>
   requires std::same_as<typename T::policy_t, framework::PreslicePolicyGeneral>
 bool registerCache(T& preslice, Cache&, Cache& bsksU)
 {
-  if constexpr (T::optional) {
-    if (preslice.binding == "[MISSING]") {
-      return true;
-    }
+  if (preslice.isMissing()) {
+    return true;
   }
   auto locate = std::find(bsksU.begin(), bsksU.end(), preslice.getBindingKey());
   if (locate == bsksU.end()) {
@@ -688,10 +751,8 @@ template <is_preslice T>
 static bool updateSliceInfo(T& preslice, ArrowTableSlicingCache& cache)
   requires std::same_as<typename T::policy_t, framework::PreslicePolicySorted>
 {
-  if constexpr (T::optional) {
-    if (preslice.binding == "[MISSING]") {
-      return true;
-    }
+  if (preslice.isMissing()) {
+    return true;
   }
   preslice.updateSliceInfo(cache.getCacheFor(preslice.getBindingKey()));
   return true;
@@ -701,10 +762,8 @@ template <is_preslice T>
 static bool updateSliceInfo(T& preslice, ArrowTableSlicingCache& cache)
   requires std::same_as<typename T::policy_t, framework::PreslicePolicyGeneral>
 {
-  if constexpr (T::optional) {
-    if (preslice.binding == "[MISSING]") {
-      return true;
-    }
+  if (preslice.isMissing()) {
+    return true;
   }
   preslice.updateSliceInfo(cache.getCacheUnsortedFor(preslice.getBindingKey()));
   return true;
diff --git a/Framework/Core/include/Framework/AnalysisSupportHelpers.h b/Framework/Core/include/Framework/AnalysisSupportHelpers.h
index c1968123e765d..6fe4b01b065b6 100644
--- a/Framework/Core/include/Framework/AnalysisSupportHelpers.h
+++ b/Framework/Core/include/Framework/AnalysisSupportHelpers.h
@@ -42,6 +42,15 @@ struct AnalysisSupportHelpers {
                                          std::vector<InputSpec>& requestedAODs,
                                          std::vector<InputSpec>& requestedDYNs,
                                          DataProcessorSpec& publisher);
+  static void addMissingOutputsToSlicer(std::vector<InputSpec> const& requestedSLCs,
+                                        DataProcessorSpec& publisher);
+  /// Split the requested slice infos into groups by the device providing the sliced table
+  /// (the AOD reader, if none of the providers has it) and create a slicer device for each
+  /// group, so that each slicer depends on a single device and does not create loops.
+  /// Each slicer is returned together with the name of its provider.
+  static std::vector<std::pair<std::string, DataProcessorSpec>> makeSlicers(std::vector<InputSpec> const& requestedSLCs,
+                                                                            std::vector<DataProcessorSpec const*> const& providers,
+                                                                            std::vector<std::vector<InputSpec>>& slicerGroups);
 
   /// Match all inputs of kind ATSK and write them to a ROOT file,
   /// one root file per originating task.
diff --git a/Framework/Core/include/Framework/AnalysisTask.h b/Framework/Core/include/Framework/AnalysisTask.h
index dcef106c563e4..0d8f1794ff9bb 100644
--- a/Framework/Core/include/Framework/AnalysisTask.h
+++ b/Framework/Core/include/Framework/AnalysisTask.h
@@ -105,14 +105,7 @@ struct AnalysisDataProcessorBuilder {
     spec.metadata.emplace_back(ConfigParamSpec{std::string{"control:"} + name, VariantType::Bool, value, {"\"\""}});
     auto matcher = DataSpecUtils::asConcreteDataMatcher(spec);
     DataSpecUtils::updateInputList(inputs, std::move(spec));
-    auto locate = std::ranges::find_if(iInfos, [&hash](auto const& info) { return info.hash == hash; });
-    if (locate == iInfos.end()) {
-      iInfos.emplace_back(hash, std::vector{std::pair{ai, matcher}});
-    } else {
-      if (std::ranges::none_of(locate->matchers, [&ai, &matcher](auto const& match) { return (match.first == ai) && (match.second == matcher); })) {
-        locate->matchers.emplace_back(std::pair{ai, matcher});
-      }
-    }
+    updateInputInfos(iInfos, std::move(matcher), hash, ai);
   }
 
   /// helpers to append expression information for a single argument
@@ -144,6 +137,38 @@ struct AnalysisDataProcessorBuilder {
     }.template operator()<A::originals.size(), std::decay_t<A>::originals>(std::make_index_sequence<std::decay_t<A>::originals.size()>());
   }
 
+  /// helpers to append InputSpect for a single argument with grouping
+  template <soa::is_iterator G, soa::is_table A>
+    requires(soa::relatedByIndex<std::decay_t<G>, std::decay_t<A>>())
+  static void addSlicingInput(const char* name, bool value, std::vector<InputSpec>& inputs, header::DataOrigin&& newOrigin = header::DataOrigin{"AOD"})
+  {
+    auto key = std::string{"fIndex"} + o2::framework::cutString(soa::getLabelFromType<std::decay_t<G>>());
+    Entry entry{soa::getLabelFromTypeForKey<std::decay_t<A>>(key), soa::getMatcherFromTypeForKey<std::decay_t<A>>(key), key, value};
+    // replace the origin of the sliced table first, so that both the slicer source and the slice info carry the new origin
+    bool originReplaced = (entry.matcher.origin == header::DataOrigin{"AOD"}) && (newOrigin != header::DataOrigin{"AOD"});
+    if (originReplaced) {
+      entry.matcher = replaceOrigin(entry.matcher, newOrigin);
+    }
+    auto spec = inputForEntry(entry, !o2::soa::is_smallgroups<std::decay_t<A>>);
+    spec.metadata.emplace_back(ConfigParamSpec{std::string{"control:"} + name, VariantType::Bool, value, {"\"\""}});
+    if (originReplaced) {
+      spec.metadata.emplace_back(ConfigParamSpec{"aod-origin-replaced", VariantType::Bool, true, {"\"\""}});
+    }
+
+    DataSpecUtils::updateInputList(inputs, std::move(spec));
+  }
+
+  template <soa::is_iterator G, soa::is_table A>
+    requires(!soa::relatedByIndex<std::decay_t<G>, std::decay_t<A>>())
+  static void addSlicingInput(const char*, bool, std::vector<InputSpec>&, header::DataOrigin&&)
+  {
+  }
+
+  template <soa::is_table G, soa::is_table A>
+  static void addSlicingInput(const char*, bool, std::vector<InputSpec>&, header::DataOrigin&&)
+  {
+  }
+
   /// helper to append the inputs and expression information for normalized arguments
   template <soa::is_table... As>
   static void addInputsAndExpressions(uint32_t hash, const char* name, bool value, std::vector<InputSpec>& inputs, std::vector<ExpressionInfo>& eInfos, std::vector<InputInfo>& iInfos, header::DataOrigin&& newOrigin = header::DataOrigin{"AOD"})
@@ -158,6 +183,17 @@ struct AnalysisDataProcessorBuilder {
      ...);
   }
 
+  /// helper to append the slicing inputs
+  template <soa::is_iterator G, soa::is_table... As>
+  static void addSlicingInputs(const char* name, bool value, std::vector<InputSpec>& inputs, header::DataOrigin&& newOrigin = header::DataOrigin{"AOD"})
+  {
+    ([&name, &value, &inputs, newOrigin]() mutable {
+      using T = std::decay_t<As>;
+      addSlicingInput<G, T>(name, value, inputs, std::move(newOrigin));
+    }(),
+     ...);
+  }
+
   /// helper to parse the process arguments
   template <typename T>
   inline static bool requestInputsFromArgs(T&, std::string const&, std::vector<InputSpec>&, std::vector<ExpressionInfo>&, std::vector<InputInfo>&, header::DataOrigin)
@@ -196,7 +232,11 @@ struct AnalysisDataProcessorBuilder {
     requires(std::is_lvalue_reference_v<A> && (std::is_lvalue_reference_v<Args> && ...))
   {
     constexpr auto hash = o2::framework::TypeIdHelpers::uniqueId<void (C::*)(A, Args...)>();
-    addInputsAndExpressions<typename std::decay_t<A>::parent_t, Args...>(hash, name, value, inputs, eInfos, iInfos, std::move(newOrigin));
+    // here we also add slicing inputs with the same constrol config param spec as normal inputs, so it can be removed in adjust topology
+    // step if the process function is disabled
+    auto newOriginCopy = newOrigin;
+    addSlicingInputs<A, Args...>(name, value, inputs, std::move(newOrigin));
+    addInputsAndExpressions<typename std::decay_t<A>::parent_t, Args...>(hash, name, value, inputs, eInfos, iInfos, std::move(newOriginCopy));
   }
 
   /// 3. generic case
@@ -595,16 +635,50 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
   // replace origins in Preslice declarations
   homogeneous_apply_refs_sized<numElements>([&newOrigin](auto& element) { return analysis_task_parsers::replaceOrigin(element, newOrigin); }, *task.get());
 
-  auto algo = AlgorithmSpec::InitCallback
-  {
+  // add slicing inputs from Preslice declarations
+  homogeneous_apply_refs_sized<numElements>(
+    [&inputs, &newOrigin](auto& element) {
+      return analysis_task_parsers::addSlicingInputs(element, inputs, newOrigin);
+    },
+    *task.get());
+
+  auto algo = AlgorithmSpec::InitCallback{
     [task = task, expressionInfos, inputInfos, newOrigin, newOriginStr](InitContext& ic) mutable {
+      homogeneous_apply_refs_sized<numElements>([&ic](auto&& element) { return analysis_task_parsers::prepareOption(ic, element); }, *task.get());
+      homogeneous_apply_refs_sized<numElements>([&ic](auto&& element) { return analysis_task_parsers::prepareService(ic, element); }, *task.get());
+
       Cache bindingsKeys;
       Cache bindingsKeysUnsorted;
       // add preslice declarations to slicing cache definition
-      homogeneous_apply_refs_sized<numElements>([&bindingsKeys, &bindingsKeysUnsorted](auto& element) { return analysis_task_parsers::registerCache(element, bindingsKeys, bindingsKeysUnsorted); }, *task.get());
+      homogeneous_apply_refs_sized<numElements>(
+        [&bindingsKeys, &bindingsKeysUnsorted](auto& element) {
+          return analysis_task_parsers::registerCache(element, bindingsKeys, bindingsKeysUnsorted);
+        },
+        *task.get());
 
-      homogeneous_apply_refs_sized<numElements>([&ic](auto&& element) { return analysis_task_parsers::prepareOption(ic, element); }, *task.get());
-      homogeneous_apply_refs_sized<numElements>([&ic](auto&& element) { return analysis_task_parsers::prepareService(ic, element); }, *task.get());
+      /// parse process functions to enable requested grouping caches - note that at this state process configurables have their final values
+      if constexpr (requires { &T::process; }) {
+        AnalysisDataProcessorBuilder::cacheFromArgs(&T::process, true, bindingsKeys, bindingsKeysUnsorted);
+      }
+      homogeneous_apply_refs_sized<numElements>(
+        [&bindingsKeys, &bindingsKeysUnsorted](auto& x) {
+          return AnalysisDataProcessorBuilder::requestCacheFromArgs(x, bindingsKeys, bindingsKeysUnsorted);
+        },
+        *task.get());
+
+      /// replace origin in slicing caches
+      std::ranges::transform(bindingsKeys, bindingsKeys.begin(), [&newOrigin](Entry& entry) {
+        if ((entry.matcher.origin == header::DataOrigin{"AOD"}) && (newOrigin != header::DataOrigin{"AOD"})) {
+          entry.matcher = replaceOrigin(entry.matcher, newOrigin);
+        }
+        return entry;
+      });
+      std::ranges::transform(bindingsKeysUnsorted, bindingsKeysUnsorted.begin(), [&newOrigin](Entry& entry) {
+        if ((entry.matcher.origin == header::DataOrigin{"AOD"}) && (newOrigin != header::DataOrigin{"AOD"})) {
+          entry.matcher = replaceOrigin(entry.matcher, newOrigin);
+        }
+        return entry;
+      });
 
       auto& callbacks = ic.services().get<CallbackService>();
       auto eoscb = [task](EndOfStreamContext& eosContext) {
@@ -633,30 +707,6 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
       },
                                                 *task.get());
 
-      /// parse process functions to enable requested grouping caches - note that at this state process configurables have their final values
-      if constexpr (requires { &T::process; }) {
-        AnalysisDataProcessorBuilder::cacheFromArgs(&T::process, true, bindingsKeys, bindingsKeysUnsorted);
-      }
-      homogeneous_apply_refs_sized<numElements>(
-        [&bindingsKeys, &bindingsKeysUnsorted](auto& x) {
-          return AnalysisDataProcessorBuilder::requestCacheFromArgs(x, bindingsKeys, bindingsKeysUnsorted);
-        },
-        *task.get());
-
-      /// replace origin in slicing caches
-      std::ranges::transform(bindingsKeys, bindingsKeys.begin(), [&newOrigin](Entry& entry) {
-        if ((entry.matcher.origin == header::DataOrigin{"AOD"}) && (newOrigin != header::DataOrigin{"AOD"})) {
-          entry.matcher = replaceOrigin(entry.matcher, newOrigin);
-        }
-        return entry;
-      });
-      std::ranges::transform(bindingsKeysUnsorted, bindingsKeysUnsorted.begin(), [&newOrigin](Entry& entry) {
-        if ((entry.matcher.origin == header::DataOrigin{"AOD"}) && (newOrigin != header::DataOrigin{"AOD"})) {
-          entry.matcher = replaceOrigin(entry.matcher, newOrigin);
-        }
-        return entry;
-      });
-
       ic.services().get<ArrowTableSlicingCacheDef>().setCaches(std::move(bindingsKeys));
       ic.services().get<ArrowTableSlicingCacheDef>().setCachesUnsorted(std::move(bindingsKeysUnsorted));
       ic.services().get<ArrowTableSlicingCacheDef>().setOrigin(newOrigin);
@@ -715,8 +765,7 @@ DataProcessorSpec adaptAnalysisTask(ConfigContext const& ctx, Args&&... args)
         // finalize outputs
         homogeneous_apply_refs_sized<numElements>([&pc](auto& element) { return analysis_task_parsers::finalizeOutput(pc, element); }, *task.get());
       };
-    }
-  };
+    }};
 
   return {
     name,
diff --git a/Framework/Core/include/Framework/ArrowTableSlicingCache.h b/Framework/Core/include/Framework/ArrowTableSlicingCache.h
index 9b27480024674..e0aab4b18c14e 100644
--- a/Framework/Core/include/Framework/ArrowTableSlicingCache.h
+++ b/Framework/Core/include/Framework/ArrowTableSlicingCache.h
@@ -13,14 +13,16 @@
 #define ARROWTABLESLICINGCACHE_H
 
 #include "Framework/ConcreteDataMatcher.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/InputSpec.h"
 #include "Framework/ServiceHandle.h"
+#include "Framework/StringHelpers.h"
 #include <arrow/array.h>
+#include <fmt/format.h>
 #include <gsl/span>
 
 namespace o2::framework
 {
-using ListVector = std::vector<std::vector<int64_t>>;
-
 struct SliceInfoPtr {
   gsl::span<int64_t const> offsets;
   gsl::span<int64_t const> sizes;
@@ -28,9 +30,10 @@ struct SliceInfoPtr {
   std::pair<int64_t, int64_t> getSliceFor(int value) const;
 };
 
+/// view of an unsorted slice-info table: rows of group v are rows[offsets[v], offsets[v + 1])
 struct SliceInfoUnsortedPtr {
-  std::span<int const> values;
-  ListVector const* groups;
+  std::span<int32_t const> offsets;
+  std::span<int64_t const> rows;
 
   std::span<int64_t const> getSliceFor(int value) const;
 };
@@ -56,6 +59,32 @@ struct Entry {
   }
 };
 
+InputSpec inputForEntry(Entry const& entry, bool sorted);
+ConcreteDataMatcher matcherForEntry(Entry const& entry);
+ConcreteDataMatcher matcherForMatcherAndKey(ConcreteDataMatcher const& matcher, std::string const& key);
+
+/// Layout of the slice-info tables produced by the internal slicer device.
+/// Row v describes the group with index value v, for v in [0, max index value];
+/// rows with negative index values do not belong to any group.
+struct SliceInfo {
+  /// sorted: group v is the contiguous range [fOffset, fOffset + fSize)
+  static constexpr const char* offsetsLabel = "fOffset"; // int64
+  static constexpr const char* sizesLabel = "fSize";     // int64
+  /// unsorted: group v is the list of row numbers in fRows
+  static constexpr const char* rowsLabel = "fRows"; // list<int64>
+
+  static std::shared_ptr<arrow::Schema> sortedSchema();
+  static std::shared_ptr<arrow::Schema> unsortedSchema();
+
+  /// build slice-info tables for the index column entry.key of the source table
+  static std::shared_ptr<arrow::Table> makeSorted(Entry const& entry, std::shared_ptr<arrow::Table> const& source);
+  static std::shared_ptr<arrow::Table> makeUnsorted(Entry const& entry, std::shared_ptr<arrow::Table> const& source);
+
+  /// non-owning views of the slice-info tables, valid as long as the table is alive
+  static SliceInfoPtr readSorted(std::shared_ptr<arrow::Table> const& table);
+  static SliceInfoUnsortedPtr readUnsorted(std::shared_ptr<arrow::Table> const& table);
+};
+
 using Cache = std::vector<Entry>;
 
 void updatePairList(Cache& list, Entry& entry);
@@ -77,13 +106,14 @@ struct ArrowTableSlicingCacheDef {
 struct ArrowTableSlicingCache {
   constexpr static ServiceKind service_kind = ServiceKind::Stream;
 
+  // slice-info tables (see SliceInfo) for the current timeframe and views into them
   Cache bindingsKeys;
-  std::vector<std::vector<int64_t>> offsets;
-  std::vector<std::vector<int64_t>> sizes;
+  std::vector<std::shared_ptr<arrow::Table>> sliceInfos;
+  std::vector<SliceInfoPtr> sliceInfoPtrs;
 
   Cache bindingsKeysUnsorted;
-  std::vector<std::vector<int>> valuesUnsorted;
-  std::vector<ListVector> groups;
+  std::vector<std::shared_ptr<arrow::Table>> sliceInfosUnsorted;
+  std::vector<SliceInfoUnsortedPtr> sliceInfoPtrsUnsorted;
 
   header::DataOrigin newOrigin = header::DataOrigin{"AOD"};
 
@@ -92,7 +122,13 @@ struct ArrowTableSlicingCache {
   // set caching information externally
   void setCaches(Cache&& bsks, Cache&& bsksUnsorted = {});
 
-  // update slicing info cache entry (assumes it is already present)
+  // store slice-info table received for the cache entry (assumes it is already present)
+  void setCacheEntry(int pos, std::shared_ptr<arrow::Table> sliceInfo);
+  void setCacheEntryUnsorted(int pos, std::shared_ptr<arrow::Table> sliceInfo);
+  // drop all slice-info tables, e.g. at the start of a new timeframe
+  void clearCacheEntries();
+
+  // compute slice-info table for the cache entry locally from the sliced table (assumes it is already present)
   arrow::Status updateCacheEntry(int pos, std::shared_ptr<arrow::Table> const& table);
   arrow::Status updateCacheEntryUnsorted(int pos, std::shared_ptr<arrow::Table> const& table);
 
@@ -107,12 +143,6 @@ struct ArrowTableSlicingCache {
   SliceInfoPtr getCacheForPos(int pos) const;
   SliceInfoUnsortedPtr getCacheUnsortedForPos(int pos) const;
 
-  // get a cached empty (0-row) slice of the given table, so that empty groups
-  // do not slice every column only to produce 0 rows (the common case for
-  // sparse grouping). One-slot cache keyed by the table pointer.
-  std::shared_ptr<arrow::Table> getEmptySliceFor(std::shared_ptr<arrow::Table> const& table);
-  std::pair<arrow::Table const*, std::shared_ptr<arrow::Table>> emptySlice{nullptr, nullptr};
-
   static void validateOrder(Entry const& bindingKey, std::shared_ptr<arrow::Table> const& input);
 };
 } // namespace o2::framework
diff --git a/Framework/Core/include/Framework/DanglingEdgesContext.h b/Framework/Core/include/Framework/DanglingEdgesContext.h
index c5f54297ee746..e000a30f9ebce 100644
--- a/Framework/Core/include/Framework/DanglingEdgesContext.h
+++ b/Framework/Core/include/Framework/DanglingEdgesContext.h
@@ -45,6 +45,10 @@ struct DanglingEdgesContext {
   // ccdb tables
   std::vector<OutputSpec> providedTIMs;
   std::vector<InputSpec> requestedTIMs;
+  // slice infos
+  std::vector<InputSpec> requestedSLCs;
+  // slice infos grouped by the device providing the sliced tables, one slicer device per group
+  std::vector<std::vector<InputSpec>> slicerGroups;
   // output objects
   std::vector<OutputSpec> providedOutputObjHist;
   // inputs for the extension spawner
diff --git a/Framework/Core/src/ASoA.cxx b/Framework/Core/src/ASoA.cxx
index cfd58ae159b7c..7df56250139bc 100644
--- a/Framework/Core/src/ASoA.cxx
+++ b/Framework/Core/src/ASoA.cxx
@@ -235,9 +235,9 @@ void notFoundColumn(const char* label, const char* key)
   throw o2::framework::runtime_error_f(R"(Preslice not valid: table "%s" (or join based on it) does not have column "%s")", label, key);
 }
 
-void missingOptionalPreslice(const char* label, const char* key)
+void missingPreslice(const char* label, const char* key)
 {
-  throw o2::framework::runtime_error_f(R"(Optional Preslice with missing binding used: table "%s" (or join based on it) does not have column "%s")", label, key);
+  throw o2::framework::runtime_error_f(R"(Preslice with missing binding used: table "%s" (or join based on it) does not have column "%s")", label, key);
 }
 
 void* extractCCDBPayload(char* payload, size_t size, TClass const* cl, const char* what)
@@ -327,7 +327,7 @@ std::string strToUpper(std::string&& str)
 
 bool PreslicePolicyBase::isMissing() const
 {
-  return binding == "[MISSING]";
+  return bindingKey.binding == "[MISSING]";
 }
 
 Entry const& PreslicePolicyBase::getBindingKey() const
diff --git a/Framework/Core/src/AnalysisHelpers.cxx b/Framework/Core/src/AnalysisHelpers.cxx
index bb359e9adcaf4..178a205a83769 100644
--- a/Framework/Core/src/AnalysisHelpers.cxx
+++ b/Framework/Core/src/AnalysisHelpers.cxx
@@ -14,6 +14,21 @@
 #include "ExpressionJSONHelpers.h"
 #include "IndexJSONHelpers.h"
 
+namespace o2::framework
+{
+void updateInputInfos(std::vector<InputInfo>& iInfos, ConcreteDataMatcher&& matcher, uint32_t hash, int ai)
+{
+  auto locate = std::ranges::find_if(iInfos, [&hash](auto const& info) { return info.hash == hash; });
+  if (locate == iInfos.end()) {
+    iInfos.emplace_back(hash, std::vector{std::pair{ai, matcher}});
+  } else {
+    if (std::ranges::none_of(locate->matchers, [&ai, &matcher](auto const& match) { return (match.first == ai) && (match.second == matcher); })) {
+      locate->matchers.emplace_back(std::pair{ai, matcher});
+    }
+  }
+}
+} // namespace o2::framework
+
 namespace o2::soa
 {
 std::vector<framework::IndexColumnBuilder> IndexBuilder::makeBuilders(std::vector<std::shared_ptr<arrow::Table>>&& tables, std::vector<soa::IndexRecord> const& records)
diff --git a/Framework/Core/src/AnalysisSupportHelpers.cxx b/Framework/Core/src/AnalysisSupportHelpers.cxx
index c16a1da61ae8a..5c4516c2f9eb7 100644
--- a/Framework/Core/src/AnalysisSupportHelpers.cxx
+++ b/Framework/Core/src/AnalysisSupportHelpers.cxx
@@ -188,6 +188,63 @@ void AnalysisSupportHelpers::addMissingOutputsToBuilder(std::vector<InputSpec> c
     sinks::update_input_list{requestedDYNs}; // update requestedDYNs
 }
 
+void AnalysisSupportHelpers::addMissingOutputsToSlicer(std::vector<InputSpec> const& requestedSLCs,
+                                                       DataProcessorSpec& publisher)
+{
+  requestedSLCs |
+    views::input_to_output_specs() |
+    sinks::append_to{publisher.outputs};
+
+  for (auto const& input : requestedSLCs) {
+    input.metadata |
+      views::filter_string_params_starts_with("slice-source:") |
+      views::params_to_input_specs() |
+      sinks::update_input_list{publisher.inputs};
+  }
+}
+
+std::vector<std::pair<std::string, DataProcessorSpec>> AnalysisSupportHelpers::makeSlicers(std::vector<InputSpec> const& requestedSLCs,
+                                                                                           std::vector<DataProcessorSpec const*> const& providers,
+                                                                                           std::vector<std::vector<InputSpec>>& slicerGroups)
+{
+  // find the device providing the sliced table, if there is none the table is read from file
+  auto providerFor = [&providers](InputSpec const& request) -> std::string {
+    auto sources = request.metadata |
+                   views::filter_string_params_starts_with("slice-source:") |
+                   views::params_to_input_specs();
+    auto matcher = DataSpecUtils::asConcreteDataMatcher(*sources.begin());
+    auto provider = std::ranges::find_if(providers, [&matcher](DataProcessorSpec const* spec) {
+      return std::ranges::any_of(spec->outputs, [&matcher](OutputSpec const& output) { return DataSpecUtils::match(output, matcher); });
+    });
+    return provider != providers.end() ? (*provider)->name : "internal-dpl-aod-reader";
+  };
+
+  slicerGroups.clear();
+  std::vector<std::string> groupProviders;
+  for (auto const& request : requestedSLCs) {
+    auto provider = providerFor(request);
+    auto locate = std::ranges::find(groupProviders, provider);
+    if (locate == groupProviders.end()) {
+      groupProviders.push_back(provider);
+      slicerGroups.push_back({request});
+    } else {
+      slicerGroups[std::distance(groupProviders.begin(), locate)].push_back(request);
+    }
+  }
+
+  std::vector<std::pair<std::string, DataProcessorSpec>> slicers;
+  for (auto i = 0u; i < slicerGroups.size(); ++i) {
+    DataProcessorSpec slicer{.name = "internal-dpl-aod-slicer-" + std::to_string(i),
+                             .inputs = {},
+                             .outputs = {},
+                             .algorithm = AlgorithmSpec::dummyAlgorithm(), // real algorithm will be set in adjustTopology
+                             .options = {ConfigParamSpec{"slicer-group", VariantType::Int, static_cast<int>(i), {"index of the slice info group handled by this slicer"}}}};
+    addMissingOutputsToSlicer(slicerGroups[i], slicer);
+    slicers.emplace_back(groupProviders[i], std::move(slicer));
+  }
+  return slicers;
+}
+
 // =============================================================================
 DataProcessorSpec AnalysisSupportHelpers::getOutputObjHistSink(ConfigContext const& ctx)
 {
diff --git a/Framework/Core/src/ArrowSupport.cxx b/Framework/Core/src/ArrowSupport.cxx
index cf72696ff7445..1832e73a97dce 100644
--- a/Framework/Core/src/ArrowSupport.cxx
+++ b/Framework/Core/src/ArrowSupport.cxx
@@ -698,6 +698,26 @@ o2::framework::ServiceSpec ArrowSupport::arrowBackendSpec()
         }
       }
 
+      // slicers are recreated from scratch, grouped by the devices providing the sliced tables
+      std::erase_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-slicer"); });
+      dec.requestedSLCs.clear();
+      for (auto& d : workflow) {
+        d.inputs |
+          views::filter_with_params_by_name_starting("slice-source:") |
+          sinks::update_input_list{dec.requestedSLCs};
+      }
+      std::ranges::sort(dec.requestedSLCs, inputSpecLessThan);
+      std::vector<DataProcessorSpec const*> slicedTablesProviders;
+      std::ranges::transform(workflow, std::back_inserter(slicedTablesProviders), [](DataProcessorSpec const& spec) { return &spec; });
+      auto slicers = AnalysisSupportHelpers::makeSlicers(dec.requestedSLCs, slicedTablesProviders, dec.slicerGroups);
+      // the slicers are placed after their providers in a pre-sorted workflow
+      for (auto& [providerName, slicer] : slicers) {
+        // load real AlgorithmSpec before deployment
+        slicer.algorithm = PluginManager::loadAlgorithmFromPlugin("O2FrameworkOnDemandTablesSupport", "ArrowTableSlicer", ctx);
+        auto provider = std::ranges::find(workflow, providerName, &DataProcessorSpec::name);
+        workflow.insert(provider == workflow.end() ? workflow.begin() : std::next(provider), std::move(slicer));
+      }
+
       auto writer = std::ranges::find_if(workflow, [](DataProcessorSpec const& spec) { return spec.name.starts_with("internal-dpl-aod-writer"); });
       if (writer != workflow.end()) {
         workflow.erase(writer);
@@ -772,23 +792,28 @@ o2::framework::ServiceSpec ArrowSupport::arrowTableSlicingCacheSpec()
                                                                                                          ServiceKind::Stream, typeid(ArrowTableSlicingCache).name()}; },
     .configure = CommonServices::noConfiguration(),
     .preProcessing = [](ProcessingContext& pc, void* service_ptr) {
+      // pick up the slice-info tables provided by the internal slicers for this timeframe
       auto* service = static_cast<ArrowTableSlicingCache*>(service_ptr);
+      service->clearCacheEntries();
       auto const& caches = service->bindingsKeys;
+      auto hasInput = [&pc](Entry const& entry, ConcreteDataMatcher const& matcher) {
+        if (pc.inputs().getPos(matcher) >= 0) {
+          return true;
+        }
+        LOGP(warn, "No slice info input {} for {}/{} ({}), its slices will be empty", DataSpecUtils::describe(matcher), entry.binding, entry.key, DataSpecUtils::describe(entry.matcher));
+        return false;
+      };
       for (auto i = 0u; i < caches.size(); ++i) {
-        if (caches[i].enabled && pc.inputs().getPos(caches[i].binding.c_str()) >= 0) {
-          auto status = service->updateCacheEntry(i, pc.inputs().get<TableConsumer>(caches[i].matcher)->asArrowTable());
-          if (!status.ok()) {
-            throw runtime_error_f("Failed to update slice cache for %s/%s", caches[i].binding.c_str(), caches[i].key.c_str());
-          }
+        auto matcher = matcherForEntry(caches[i]);
+        if (caches[i].enabled && hasInput(caches[i], matcher)) {
+          service->setCacheEntry(i, pc.inputs().get<TableConsumer>(matcher)->asArrowTable());
         }
       }
       auto const& unsortedCaches = service->bindingsKeysUnsorted;
       for (auto i = 0u; i < unsortedCaches.size(); ++i) {
-        if (unsortedCaches[i].enabled && pc.inputs().getPos(unsortedCaches[i].binding.c_str()) >= 0) {
-          auto status = service->updateCacheEntryUnsorted(i, pc.inputs().get<TableConsumer>(unsortedCaches[i].matcher)->asArrowTable());
-          if (!status.ok()) {
-            throw runtime_error_f("failed to update slice cache (unsorted) for %s/%s", unsortedCaches[i].binding.c_str(), unsortedCaches[i].key.c_str());
-          }
+        auto matcher = matcherForEntry(unsortedCaches[i]);
+        if (unsortedCaches[i].enabled && hasInput(unsortedCaches[i], matcher)) {
+          service->setCacheEntryUnsorted(i, pc.inputs().get<TableConsumer>(matcher)->asArrowTable());
         }
       } },
     .kind = ServiceKind::Stream};
diff --git a/Framework/Core/src/ArrowTableSlicingCache.cxx b/Framework/Core/src/ArrowTableSlicingCache.cxx
index 7b96b7c0c0e85..1f402e870a2a5 100644
--- a/Framework/Core/src/ArrowTableSlicingCache.cxx
+++ b/Framework/Core/src/ArrowTableSlicingCache.cxx
@@ -17,6 +17,8 @@
 #include <arrow/compute/kernel.h>
 #include <arrow/table.h>
 
+#include <numeric>
+
 namespace o2::framework
 {
 
@@ -44,8 +46,92 @@ arrow::ChunkedArray* getIndexFromLabel(arrow::Table* table, std::string_view lab
   }
   return table->column(std::distance(table->schema()->fields().begin(), field)).get();
 }
+
+// collect offset and size of each group of a sorted index column
+void fillSorted(arrow::ChunkedArray* column, std::vector<int64_t>& offsets, std::vector<int64_t>& sizes)
+{
+  int maxValue = -1;
+  // starting from the end, find the first positive value, in a sorted column it is the largest index
+  for (auto iChunk = column->num_chunks() - 1; iChunk >= 0; --iChunk) {
+    auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
+    for (auto iElement = chunk.length() - 1; iElement >= 0; --iElement) {
+      auto value = chunk.Value(iElement);
+      if (value < 0) {
+        continue;
+      } else {
+        maxValue = value;
+        break;
+      }
+    }
+    if (maxValue >= 0) {
+      break;
+    }
+  }
+
+  offsets.resize(maxValue + 1);
+  sizes.resize(maxValue + 1);
+
+  // loop over the index and collect size/offset
+  int lastValue = std::numeric_limits<int>::max();
+  int globalRow = 0;
+  for (auto iChunk = 0; iChunk < column->num_chunks(); ++iChunk) {
+    auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
+    for (auto iElement = 0; iElement < chunk.length(); ++iElement) {
+      auto v = chunk.Value(iElement);
+      if (v >= 0) {
+        if (v == lastValue) {
+          ++sizes[v];
+        } else {
+          lastValue = v;
+          ++sizes[v];
+          offsets[v] = globalRow;
+        }
+      }
+      ++globalRow;
+    }
+  }
+}
 } // namespace
 
+InputSpec inputForEntry(Entry const& entry, bool sorted)
+{
+  // the slice info table inherits the sliced table binding and origin, while using hash
+  // of original description and normalized column name as a new description
+  auto& [origin, description, version] = entry.matcher;
+  auto newdescription = std::string{description.str} + "/" + entry.key;
+  auto hash = runtime_hash(newdescription.c_str());
+  auto d = header::DataDescription{"initial"};
+  d.runtimeInit(std::to_string(hash).c_str());
+  InputSpec result{entry.binding + "_Slice", origin, d, version};
+  // add metadata to retrieve the original table
+  result.metadata.emplace_back(
+    o2::framework::ConfigParamSpec{fmt::format("slice-source:{}", entry.binding),
+                                   framework::VariantType::String,
+                                   fmt::format("{}/{}/{}/{}", entry.binding, origin.as<std::string>(), description.as<std::string>(), version),
+                                   {"\"\""}});
+  result.metadata.emplace_back(
+    o2::framework::ConfigParamSpec{"slice-key", framework::VariantType::String, entry.key, {"\"\""}});
+  result.metadata.emplace_back(
+    o2::framework::ConfigParamSpec{"sorted", framework::VariantType::Bool, sorted, {"\"\""}});
+
+  return result;
+}
+
+ConcreteDataMatcher matcherForEntry(Entry const& entry)
+{
+  return matcherForMatcherAndKey(entry.matcher, entry.key);
+}
+
+ConcreteDataMatcher matcherForMatcherAndKey(ConcreteDataMatcher const& matcher, std::string const& key)
+{
+  auto& [origin, description, version] = matcher;
+  auto newdescription = std::string{description.str} + "/" + key;
+  auto hash = runtime_hash(newdescription.c_str());
+  auto d = header::DataDescription{"initial"};
+  d.runtimeInit(std::to_string(hash).c_str());
+  return {origin, d, version};
+}
+
 void updatePairList(Cache& list, Entry& entry)
 {
   auto locate = std::find(list.begin(), list.end(), entry);
@@ -65,16 +151,109 @@ std::pair<int64_t, int64_t> SliceInfoPtr::getSliceFor(int value) const
   return {offsets[value], sizes[value]};
 }
 
-std::span<const int64_t> SliceInfoUnsortedPtr::getSliceFor(int value) const
+std::span<int64_t const> SliceInfoUnsortedPtr::getSliceFor(int value) const
 {
-  if (values.empty()) {
+  if (value < 0 || (size_t)value + 1 >= offsets.size()) {
     return {};
   }
-  if (value > values[values.size() - 1]) {
+  return rows.subspan(offsets[value], offsets[value + 1] - offsets[value]);
+}
+
+std::shared_ptr<arrow::Schema> SliceInfo::sortedSchema()
+{
+  return arrow::schema({arrow::field(offsetsLabel, arrow::int64()), arrow::field(sizesLabel, arrow::int64())});
+}
+
+std::shared_ptr<arrow::Schema> SliceInfo::unsortedSchema()
+{
+  return arrow::schema({arrow::field(rowsLabel, arrow::list(arrow::int64()))});
+}
+
+std::shared_ptr<arrow::Table> SliceInfo::makeSorted(Entry const& entry, std::shared_ptr<arrow::Table> const& source)
+{
+  std::vector<int64_t> offsets;
+  std::vector<int64_t> sizes;
+  if (source->num_rows() != 0) {
+    ArrowTableSlicingCache::validateOrder(entry, source);
+    fillSorted(getIndexFromLabel(source.get(), entry.key), offsets, sizes);
+  }
+  auto length = static_cast<int64_t>(offsets.size());
+  return arrow::Table::Make(sortedSchema(),
+                            {std::make_shared<arrow::Int64Array>(length, arrow::Buffer::FromVector(std::move(offsets))),
+                             std::make_shared<arrow::Int64Array>(length, arrow::Buffer::FromVector(std::move(sizes)))},
+                            length);
+}
+
+std::shared_ptr<arrow::Table> SliceInfo::makeUnsorted(Entry const& entry, std::shared_ptr<arrow::Table> const& source)
+{
+  std::vector<int32_t> offsets{0};
+  std::vector<int64_t> rows;
+  if (source->num_rows() != 0) {
+    auto column = getIndexFromLabel(source.get(), entry.key);
+    // count the rows in each group
+    std::vector<int32_t> counts;
+    for (auto iChunk = 0; iChunk < column->num_chunks(); ++iChunk) {
+      auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
+      for (auto iElement = 0; iElement < chunk.length(); ++iElement) {
+        auto v = chunk.Value(iElement);
+        if (v >= 0) {
+          if ((int)counts.size() <= v) {
+            counts.resize(v + 1);
+          }
+          ++counts[v];
+        }
+      }
+    }
+    offsets.resize(counts.size() + 1);
+    std::inclusive_scan(counts.begin(), counts.end(), offsets.begin() + 1);
+    rows.resize(offsets.back());
+
+    // place the row numbers of each group, reusing counts as fill positions
+    std::copy(offsets.begin(), offsets.end() - 1, counts.begin());
+    int64_t row = 0;
+    for (auto iChunk = 0; iChunk < column->num_chunks(); ++iChunk) {
+      auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
+      for (auto iElement = 0; iElement < chunk.length(); ++iElement) {
+        auto v = chunk.Value(iElement);
+        if (v >= 0) {
+          rows[counts[v]++] = row;
+        }
+        ++row;
+      }
+    }
+  }
+  auto length = static_cast<int64_t>(offsets.size()) - 1;
+  auto nRows = static_cast<int64_t>(rows.size());
+  auto values = std::make_shared<arrow::Int64Array>(nRows, arrow::Buffer::FromVector(std::move(rows)));
+  return arrow::Table::Make(unsortedSchema(),
+                            {std::make_shared<arrow::ListArray>(arrow::list(arrow::int64()), length, arrow::Buffer::FromVector(std::move(offsets)), values)},
+                            length);
+}
+
+SliceInfoPtr SliceInfo::readSorted(std::shared_ptr<arrow::Table> const& table)
+{
+  if (table->num_rows() == 0) {
     return {};
   }
+  auto offsets = std::static_pointer_cast<arrow::Int64Array>(table->column(0)->chunk(0));
+  auto sizes = std::static_pointer_cast<arrow::Int64Array>(table->column(1)->chunk(0));
+  return {
+    gsl::span{offsets->raw_values(), (size_t)offsets->length()}, //
+    gsl::span{sizes->raw_values(), (size_t)sizes->length()}      //
+  };
+}
 
-  return {(*groups)[value].data(), (*groups)[value].size()};
+SliceInfoUnsortedPtr SliceInfo::readUnsorted(std::shared_ptr<arrow::Table> const& table)
+{
+  if (table->num_rows() == 0) {
+    return {};
+  }
+  auto list = std::static_pointer_cast<arrow::ListArray>(table->column(0)->chunk(0));
+  auto values = std::static_pointer_cast<arrow::Int64Array>(list->values());
+  return {
+    {list->raw_value_offsets(), (size_t)list->length() + 1}, //
+    {values->raw_values(), (size_t)values->length()}         //
+  };
 }
 
 void ArrowTableSlicingCacheDef::setCaches(Cache&& bsks)
@@ -92,116 +271,53 @@ ArrowTableSlicingCache::ArrowTableSlicingCache(Cache&& bsks, Cache&& bsksUnsorte
     bindingsKeysUnsorted{bsksUnsorted},
     newOrigin{newOrigin_}
 {
-  offsets.resize(bindingsKeys.size());
-  sizes.resize(bindingsKeys.size());
-
-  valuesUnsorted.resize(bindingsKeysUnsorted.size());
-  groups.resize(bindingsKeysUnsorted.size());
+  clearCacheEntries();
 }
 
 void ArrowTableSlicingCache::setCaches(Cache&& bsks, Cache&& bsksUnsorted)
 {
   bindingsKeys = bsks;
   bindingsKeysUnsorted = bsksUnsorted;
-  offsets.clear();
-  offsets.resize(bindingsKeys.size());
-  sizes.clear();
-  sizes.resize(bindingsKeys.size());
-  valuesUnsorted.clear();
-  valuesUnsorted.resize(bindingsKeysUnsorted.size());
-  groups.clear();
-  groups.resize(bindingsKeysUnsorted.size());
+  clearCacheEntries();
+}
+
+void ArrowTableSlicingCache::clearCacheEntries()
+{
+  sliceInfos.assign(bindingsKeys.size(), nullptr);
+  sliceInfoPtrs.assign(bindingsKeys.size(), {});
+  sliceInfosUnsorted.assign(bindingsKeysUnsorted.size(), nullptr);
+  sliceInfoPtrsUnsorted.assign(bindingsKeysUnsorted.size(), {});
+}
+
+void ArrowTableSlicingCache::setCacheEntry(int pos, std::shared_ptr<arrow::Table> sliceInfo)
+{
+  sliceInfoPtrs[pos] = SliceInfo::readSorted(sliceInfo);
+  sliceInfos[pos] = std::move(sliceInfo);
+}
+
+void ArrowTableSlicingCache::setCacheEntryUnsorted(int pos, std::shared_ptr<arrow::Table> sliceInfo)
+{
+  sliceInfoPtrsUnsorted[pos] = SliceInfo::readUnsorted(sliceInfo);
+  sliceInfosUnsorted[pos] = std::move(sliceInfo);
 }
 
 arrow::Status ArrowTableSlicingCache::updateCacheEntry(int pos, std::shared_ptr<arrow::Table> const& table)
 {
-  offsets[pos].clear();
-  sizes[pos].clear();
-  if (table->num_rows() == 0) {
-    return arrow::Status::OK();
-  }
   auto& [b, m, k, e] = bindingsKeys[pos];
   if (!e) {
     throw runtime_error_f("Disabled cache (%s) %s/%s update requested", DataSpecUtils::describe(m).c_str(), b.c_str(), k.c_str());
   }
-  validateOrder(bindingsKeys[pos], table);
-
-  int maxValue = -1;
-  auto column = getIndexFromLabel(table.get(), k);
-
-  // starting from the end, find the first positive value, in a sorted column it is the largest index
-  for (auto iChunk = column->num_chunks() - 1; iChunk >= 0; --iChunk) {
-    auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
-    for (auto iElement = chunk.length() - 1; iElement >= 0; --iElement) {
-      auto value = chunk.Value(iElement);
-      if (value < 0) {
-        continue;
-      } else {
-        maxValue = value;
-        break;
-      }
-    }
-    if (maxValue >= 0) {
-      break;
-    }
-  }
-
-  offsets[pos].resize(maxValue + 1);
-  sizes[pos].resize(maxValue + 1);
-
-  // loop over the index and collect size/offset
-  int lastValue = std::numeric_limits<int>::max();
-  int globalRow = 0;
-  for (auto iChunk = 0; iChunk < column->num_chunks(); ++iChunk) {
-    auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
-    for (auto iElement = 0; iElement < chunk.length(); ++iElement) {
-      auto v = chunk.Value(iElement);
-      if (v >= 0) {
-        if (v == lastValue) {
-          ++sizes[pos][v];
-        } else {
-          lastValue = v;
-          ++sizes[pos][v];
-          offsets[pos][v] = globalRow;
-        }
-      }
-      ++globalRow;
-    }
-  }
-
+  setCacheEntry(pos, SliceInfo::makeSorted(bindingsKeys[pos], table));
   return arrow::Status::OK();
 }
 
 arrow::Status ArrowTableSlicingCache::updateCacheEntryUnsorted(int pos, std::shared_ptr<arrow::Table> const& table)
 {
-  valuesUnsorted[pos].clear();
-  groups[pos].clear();
-  if (table->num_rows() == 0) {
-    return arrow::Status::OK();
-  }
   auto& [b, m, k, e] = bindingsKeysUnsorted[pos];
   if (!e) {
-    throw runtime_error_f("Disabled unsorted cache %s/%s update requested", b.c_str(), k.c_str());
+    throw runtime_error_f("Disabled unsorted cache (%s) %s/%s update requested", DataSpecUtils::describe(m).c_str(), b.c_str(), k.c_str());
   }
-  auto column = getIndexFromLabel(table.get(), k);
-  auto row = 0;
-  for (auto iChunk = 0; iChunk < column->num_chunks(); ++iChunk) {
-    auto chunk = static_cast<arrow::NumericArray<arrow::Int32Type>>(column->chunk(iChunk)->data());
-    for (auto iElement = 0; iElement < chunk.length(); ++iElement) {
-      auto v = chunk.Value(iElement);
-      if (v >= 0) {
-        if (std::find(valuesUnsorted[pos].begin(), valuesUnsorted[pos].end(), v) == valuesUnsorted[pos].end()) {
-          valuesUnsorted[pos].push_back(v);
-        }
-        if ((int)groups[pos].size() <= v) {
-          groups[pos].resize(v + 1);
-        }
-        (groups[pos])[v].push_back(row);
-      }
-      ++row;
-    }
-  }
-  std::sort(valuesUnsorted[pos].begin(), valuesUnsorted[pos].end());
+  setCacheEntryUnsorted(pos, SliceInfo::makeUnsorted(bindingsKeysUnsorted[pos], table));
   return arrow::Status::OK();
 }
 
@@ -239,10 +355,10 @@ SliceInfoPtr ArrowTableSlicingCache::getCacheFor(Entry const& bindingKey) const
 {
   auto [p, s] = getCachePos(bindingKey);
   if (!s) {
-    throw runtime_error_f("%s/%s is found in unsorted cache", bindingKey.binding.c_str(), bindingKey.key.c_str());
+    throw runtime_error_f("(%s) %s/%s is found in unsorted cache", DataSpecUtils::describe(bindingKey.matcher).c_str(), bindingKey.binding.c_str(), bindingKey.key.c_str());
   }
   if (!bindingsKeys[p].enabled) {
-    throw runtime_error_f("Disabled cache %s/%s is requested", bindingKey.binding.c_str(), bindingKey.key.c_str());
+    throw runtime_error_f("Disabled cache (%s) %s/%s is requested", DataSpecUtils::describe(bindingKey.matcher).c_str(), bindingKey.binding.c_str(), bindingKey.key.c_str());
   }
 
   return getCacheForPos(p);
@@ -263,26 +379,12 @@ SliceInfoUnsortedPtr ArrowTableSlicingCache::getCacheUnsortedFor(const Entry& bi
 
 SliceInfoPtr ArrowTableSlicingCache::getCacheForPos(int pos) const
 {
-  return {
-    gsl::span{offsets[pos].data(), offsets[pos].size()}, //
-    gsl::span(sizes[pos].data(), sizes[pos].size())      //
-  };
+  return sliceInfoPtrs[pos];
 }
 
 SliceInfoUnsortedPtr ArrowTableSlicingCache::getCacheUnsortedForPos(int pos) const
 {
-  return {
-    {reinterpret_cast<int const*>(valuesUnsorted[pos].data()), valuesUnsorted[pos].size()},
-    &(groups[pos]) //
-  };
-}
-
-std::shared_ptr<arrow::Table> ArrowTableSlicingCache::getEmptySliceFor(std::shared_ptr<arrow::Table> const& table)
-{
-  if (emptySlice.first != table.get()) {
-    emptySlice = {table.get(), table->Slice(0, 0)};
-  }
-  return emptySlice.second;
+  return sliceInfoPtrsUnsorted[pos];
 }
 
 void ArrowTableSlicingCache::validateOrder(Entry const& bindingKey, const std::shared_ptr<arrow::Table>& input)
diff --git a/Framework/Core/src/WorkflowHelpers.cxx b/Framework/Core/src/WorkflowHelpers.cxx
index 188b6653c6a43..fc98095341799 100644
--- a/Framework/Core/src/WorkflowHelpers.cxx
+++ b/Framework/Core/src/WorkflowHelpers.cxx
@@ -282,9 +282,11 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
       bool hasProjectors = false;
       bool hasIndexRecords = false;
       bool hasCCDBURLs = false;
+      bool hasSliceSource = false;
       bool wasAOD = false;
-      // all three options are exclusive
+      // all options are exclusive
       for (auto const& p : input.metadata) {
+        // wasAOD can be true or false for all of the options
         if (p.name.starts_with("aod-origin-replaced")) {
           wasAOD = true;
         }
@@ -300,6 +302,10 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
           hasCCDBURLs = true;
           break;
         }
+        if (p.name.starts_with("slice-source")) {
+          hasSliceSource = true;
+          break;
+        }
       }
       switch (input.lifetime) {
         case Lifetime::Timer: {
@@ -346,6 +352,8 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
         DataSpecUtils::updateInputList(dec.requestedIDXs, InputSpec{input});
       } else if (hasCCDBURLs) {
         DataSpecUtils::updateInputList(dec.requestedTIMs, InputSpec{input});
+      } else if (hasSliceSource) {
+        DataSpecUtils::updateInputList(dec.requestedSLCs, InputSpec{input});
       } else if (DataSpecUtils::partialMatch(input, AODOrigins) || wasAOD) {
         DataSpecUtils::updateInputList(dec.requestedAODs, InputSpec{input});
       }
@@ -358,8 +366,11 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
       bool hasIndexRecords = false;
       bool hasCCDBURLs = false;
       bool wasAOD = false;
-      // all three options are exclusive
+      // all options are exclusive
+      // provided slice outputs are ignored, they can only come from the slicer device
+      // that will be re-added in adjust topology
       for (auto const& p : output.metadata) {
+        // wasAOD can be true or false for all of the options
         if (p.name.starts_with("aod-origin-replaced")) {
           wasAOD = true;
         }
@@ -439,6 +450,13 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
   std::ranges::sort(providedCCDBs, outputSpecLessThan);
   AnalysisSupportHelpers::addMissingOutputsToReader(providedCCDBs, requestedCCDBs, ccdbBackend);
 
+  // slicers are grouped by the devices providing the sliced tables
+  std::ranges::sort(dec.requestedSLCs, inputSpecLessThan);
+  std::vector<DataProcessorSpec const*> slicedTablesProviders;
+  std::ranges::transform(workflow, std::back_inserter(slicedTablesProviders), [](DataProcessorSpec const& spec) { return &spec; });
+  slicedTablesProviders.insert(slicedTablesProviders.end(), {&aodSpawner, &indexBuilder, &aodReader});
+  auto aodSlicers = AnalysisSupportHelpers::makeSlicers(dec.requestedSLCs, slicedTablesProviders, dec.slicerGroups);
+
   std::vector<DataProcessorSpec> extraSpecs;
 
   if (transientStore.outputs.empty() == false) {
@@ -456,6 +474,9 @@ void WorkflowHelpers::injectServiceDevices(WorkflowSpec& workflow, ConfigContext
     extraSpecs.push_back(indexBuilder);
   }
 
+  // here the slicers are just added, unlike in adjustTopology
+  std::ranges::transform(aodSlicers, std::back_inserter(extraSpecs), [](auto&& pair) { return pair.second; });
+
   // add the Analysys CCDB backend which reads CCDB objects using a provided table
   DeploymentMode deploymentMode = DefaultsHelpers::deploymentMode();
   if (deploymentMode != DeploymentMode::OnlineDDS && deploymentMode != DeploymentMode::OnlineECS) {
diff --git a/Framework/Core/test/test_AnalysisTask.cxx b/Framework/Core/test/test_AnalysisTask.cxx
index cb710b9a3871c..3988d0ab6b61f 100644
--- a/Framework/Core/test/test_AnalysisTask.cxx
+++ b/Framework/Core/test/test_AnalysisTask.cxx
@@ -177,10 +177,10 @@ struct KTask {
   std::shared_ptr<int> someSharedInt;
 };
 
+// PresliceOptional ignores situations where the table is present, but does
+// not have the requested column
 struct LTask {
   SliceCache cache;
-  Preslice<aod::Tracks> perCol = aod::track::collisionId;
-  PresliceOptional<aod::Tracks> perPart = aod::mctracklabel::mcParticleId;
   PresliceUnsorted<aod::McCollisionLabels> perMcCol = aod::mccollisionlabel::mcCollisionId;
   PresliceUnsortedOptional<aod::Collisions> perMcColopt = aod::mccollisionlabel::mcCollisionId;
   void process(aod::McCollision const&, soa::SmallGroups<soa::Join<aod::Collisions, aod::McCollisionLabels>> const&) {}
@@ -189,14 +189,31 @@ struct LTask {
 struct MTask {
   SliceCache cache;
   struct : public PresliceGroup {
-    Preslice<aod::Tracks> perCol = aod::track::collisionId;
-    PresliceOptional<aod::Tracks> perPart = aod::mctracklabel::mcParticleId;
     PresliceUnsorted<aod::McCollisionLabels> perMcCol = aod::mccollisionlabel::mcCollisionId;
     PresliceUnsortedOptional<aod::Collisions> perMcColopt = aod::mccollisionlabel::mcCollisionId;
   } foo;
   void process(aod::McCollision const&, soa::SmallGroups<soa::Join<aod::Collisions, aod::McCollisionLabels>> const&) {}
 };
 
+// Preslices that cannot be used are skipped with a warning:
+// - table is not an input and does not have the column
+// - table is not an input
+// - table does not have the column
+struct NTask {
+  SliceCache cache;
+  Preslice<aod::Tracks> perPart = aod::mctracklabel::mcParticleId;
+  Preslice<aod::Tracks> perCol = aod::track::collisionId;
+  Preslice<aod::Collisions> perMcCol = aod::mccollisionlabel::mcCollisionId;
+  void process(aod::McCollision const&, soa::SmallGroups<soa::Join<aod::Collisions, aod::McCollisionLabels>> const&) {}
+};
+
+// optional Preslice on a table that is not an input of any process function is skipped silently
+struct OTask {
+  SliceCache cache;
+  PresliceOptional<aod::Tracks> perCol = aod::track::collisionId;
+  void process(aod::McCollision const&, soa::SmallGroups<soa::Join<aod::Collisions, aod::McCollisionLabels>> const&) {}
+};
+
 TEST_CASE("AdaptorCompilation")
 {
   auto cfgc = makeEmptyConfigContext();
@@ -218,23 +235,25 @@ TEST_CASE("AdaptorCompilation")
   REQUIRE(task1ngc.inputs[4].binding == "B3s");
 
   auto task2 = adaptAnalysisTask<BTask>(*cfgc, TaskName{"test2"});
-  REQUIRE(task2.inputs.size() == 10);
-  REQUIRE(task2.inputs[2].binding == "TracksExtension");
-  REQUIRE(task2.inputs[1].binding == "Tracks");
-  REQUIRE(task2.inputs[4].binding == "TracksExtra_002Extension");
-  REQUIRE(task2.inputs[3].binding == "TracksExtra");
-  REQUIRE(task2.inputs[6].binding == "TracksCovExtension");
-  REQUIRE(task2.inputs[5].binding == "TracksCov");
-  REQUIRE(task2.inputs[7].binding == "AmbiguousTracks");
-  REQUIRE(task2.inputs[8].binding == "Calos");
-  REQUIRE(task2.inputs[9].binding == "CaloTriggers");
-  REQUIRE(task2.inputs[0].binding == "Collisions_001");
+  REQUIRE(task2.inputs.size() == 11); // 10 base + 1 slice
+  REQUIRE(task2.inputs[0].binding == "Tracks_Slice");
+  REQUIRE(task2.inputs[1].binding == "Collisions_001");
+  REQUIRE(task2.inputs[2].binding == "Tracks");
+  REQUIRE(task2.inputs[3].binding == "TracksExtension");
+  REQUIRE(task2.inputs[4].binding == "TracksExtra");
+  REQUIRE(task2.inputs[5].binding == "TracksExtra_002Extension");
+  REQUIRE(task2.inputs[6].binding == "TracksCov");
+  REQUIRE(task2.inputs[7].binding == "TracksCovExtension");
+  REQUIRE(task2.inputs[8].binding == "AmbiguousTracks");
+  REQUIRE(task2.inputs[9].binding == "Calos");
+  REQUIRE(task2.inputs[10].binding == "CaloTriggers");
 
   auto task3 = adaptAnalysisTask<CTask>(*cfgc, TaskName{"test3"});
-  REQUIRE(task3.inputs.size() == 3);
-  REQUIRE(task3.inputs[0].binding == "Collisions_001");
-  REQUIRE(task3.inputs[1].binding == "Tracks");
-  REQUIRE(task3.inputs[2].binding == "TracksExtension");
+  REQUIRE(task3.inputs.size() == 4); // 3 base + 1 slice
+  REQUIRE(task3.inputs[0].binding == "Tracks_Slice");
+  REQUIRE(task3.inputs[1].binding == "Collisions_001");
+  REQUIRE(task3.inputs[2].binding == "Tracks");
+  REQUIRE(task3.inputs[3].binding == "TracksExtension");
 
   auto task4 = adaptAnalysisTask<DTask>(*cfgc, TaskName{"test4"});
   REQUIRE(task4.inputs.size() == 2);
@@ -269,10 +288,16 @@ TEST_CASE("AdaptorCompilation")
   REQUIRE(task11.inputs.size() == 1);
 
   auto task12 = adaptAnalysisTask<LTask>(*cfgc, TaskName{"test12"});
-  REQUIRE(task12.inputs.size() == 3);
+  REQUIRE(task12.inputs.size() == 4); // 3 base + 1 actual slice
 
   auto task13 = adaptAnalysisTask<MTask>(*cfgc, TaskName{"test13"});
-  REQUIRE(task13.inputs.size() == 3);
+  REQUIRE(task13.inputs.size() == 4); // 3 base + 1 actual slice
+
+  auto task14 = adaptAnalysisTask<NTask>(*cfgc, TaskName{"test14"});
+  REQUIRE(task14.inputs.size() == 4); // 3 base + 1 actual slice, all Preslices skipped
+
+  auto task15 = adaptAnalysisTask<OTask>(*cfgc, TaskName{"test15"});
+  REQUIRE(task15.inputs.size() == 4); // 3 base + 1 actual slice, optional Preslice skipped
 }
 
 TEST_CASE("TestPartitionIteration")