diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
index 64ef5ef17e..b4eb848ec6 100644
--- a/beacon_node/beacon_chain/src/beacon_chain.rs
+++ b/beacon_node/beacon_chain/src/beacon_chain.rs
@@ -127,7 +127,7 @@ use tokio_stream::Stream;
 use tracing::{debug, error, info, trace, warn};
 use tree_hash::TreeHash;
 use types::blob_sidecar::FixedBlobSidecarList;
-use types::data_column_sidecar::{ColumnIndex, DataColumnIdentifier};
+use types::data_column_sidecar::ColumnIndex;
 use types::payload::BlockProductionVersion;
 use types::*;
 
@@ -1106,23 +1106,25 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map_or_else(|| self.get_blobs(block_root), Ok)
     }
 
-    pub fn get_data_column_checking_all_caches(
+    pub fn get_data_columns_checking_all_caches(
         &self,
         block_root: Hash256,
-        index: ColumnIndex,
-    ) -> Result<Option<Arc<DataColumnSidecar<T::EthSpec>>>, Error> {
-        if let Some(column) = self
+        indices: &[ColumnIndex],
+    ) -> Result<DataColumnSidecarList<T::EthSpec>, Error> {
+        let all_cached_columns_opt = self
             .data_availability_checker
-            .get_data_column(&DataColumnIdentifier { block_root, index })?
-        {
-            return Ok(Some(column));
-        }
+            .get_data_columns(block_root)
+            .or_else(|| self.early_attester_cache.get_data_columns(block_root));
 
-        if let Some(columns) = self.early_attester_cache.get_data_columns(block_root) {
-            return Ok(columns.iter().find(|c| c.index == index).cloned());
+        if let Some(mut all_cached_columns) = all_cached_columns_opt {
+            all_cached_columns.retain(|col| indices.contains(&col.index));
+            Ok(all_cached_columns)
+        } else {
+            indices
+                .iter()
+                .filter_map(|index| self.get_data_column(&block_root, index).transpose())
+                .collect::<Result<_, _>>()
         }
-
-        self.get_data_column(&block_root, &index)
     }
 
     /// Returns the block at the given root, if any.
diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs
index 033b472da0..6f292f3551 100644
--- a/beacon_node/beacon_chain/src/data_availability_checker.rs
+++ b/beacon_node/beacon_chain/src/data_availability_checker.rs
@@ -17,8 +17,8 @@ use task_executor::TaskExecutor;
 use tracing::{debug, error, info_span, Instrument};
 use types::blob_sidecar::{BlobIdentifier, BlobSidecar, FixedBlobSidecarList};
 use types::{
-    BlobSidecarList, ChainSpec, DataColumnIdentifier, DataColumnSidecar, DataColumnSidecarList,
-    Epoch, EthSpec, Hash256, RuntimeVariableList, SignedBeaconBlock,
+    BlobSidecarList, ChainSpec, DataColumnSidecarList, Epoch, EthSpec, Hash256,
+    RuntimeVariableList, SignedBeaconBlock,
 };
 
 mod error;
@@ -163,12 +163,12 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         self.availability_cache.peek_blob(blob_id)
     }
 
-    /// Get a data column from the availability cache.
-    pub fn get_data_column(
+    /// Get data columns for a block from the availability cache.
+    pub fn get_data_columns(
         &self,
-        data_column_id: &DataColumnIdentifier,
-    ) -> Result<Option<Arc<DataColumnSidecar<T::EthSpec>>>, AvailabilityCheckError> {
-        self.availability_cache.peek_data_column(data_column_id)
+        block_root: Hash256,
+    ) -> Option<DataColumnSidecarList<T::EthSpec>> {
+        self.availability_cache.peek_data_columns(block_root)
     }
 
     /// Put a list of blobs received via RPC into the availability cache. This performs KZG
diff --git a/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs b/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
index 5b5a6fcc0d..3478c183f3 100644
--- a/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
+++ b/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
@@ -16,7 +16,7 @@ use std::sync::Arc;
 use tracing::debug;
 use types::blob_sidecar::BlobIdentifier;
 use types::{
-    BlobSidecar, ChainSpec, ColumnIndex, DataColumnIdentifier, DataColumnSidecar, Epoch, EthSpec,
+    BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, Epoch, EthSpec,
     Hash256, RuntimeFixedVector, RuntimeVariableList, SignedBeaconBlock,
 };
 
@@ -404,20 +404,21 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         }
     }
 
-    /// Fetch a data column from the cache without affecting the LRU ordering
-    pub fn peek_data_column(
+    /// Fetch data columns of a given `block_root` from the cache without affecting the LRU ordering
+    pub fn peek_data_columns(
         &self,
-        data_column_id: &DataColumnIdentifier,
-    ) -> Result<Option<Arc<DataColumnSidecar<T::EthSpec>>>, AvailabilityCheckError> {
-        if let Some(pending_components) = self.critical.read().peek(&data_column_id.block_root) {
-            Ok(pending_components
-                .verified_data_columns
-                .iter()
-                .find(|data_column| data_column.as_data_column().index == data_column_id.index)
-                .map(|data_column| data_column.clone_arc()))
-        } else {
-            Ok(None)
-        }
+        block_root: Hash256,
+    ) -> Option<DataColumnSidecarList<T::EthSpec>> {
+        self.critical
+            .read()
+            .peek(&block_root)
+            .map(|pending_components| {
+                pending_components
+                    .verified_data_columns
+                    .iter()
+                    .map(|col| col.clone_arc())
+                    .collect()
+            })
     }
 
     pub fn peek_pending_components<R, F: FnOnce(Option<&PendingComponents<T::EthSpec>>) -> R>(
diff --git a/beacon_node/beacon_chain/src/data_column_verification.rs b/beacon_node/beacon_chain/src/data_column_verification.rs
index 20b5c9aa02..d11c112812 100644
--- a/beacon_node/beacon_chain/src/data_column_verification.rs
+++ b/beacon_node/beacon_chain/src/data_column_verification.rs
@@ -16,7 +16,7 @@ use std::iter;
 use std::marker::PhantomData;
 use std::sync::Arc;
 use tracing::debug;
-use types::data_column_sidecar::{ColumnIndex, DataColumnIdentifier};
+use types::data_column_sidecar::ColumnIndex;
 use types::{
     BeaconStateError, ChainSpec, DataColumnSidecar, DataColumnSubnetId, EthSpec, Hash256,
     RuntimeVariableList, SignedBeaconBlockHeader, Slot,
@@ -200,13 +200,6 @@ impl<T: BeaconChainTypes, O: ObservationStrategy> GossipVerifiedDataColumn<T, O>
         )
     }
 
-    pub fn id(&self) -> DataColumnIdentifier {
-        DataColumnIdentifier {
-            block_root: self.block_root,
-            index: self.data_column.index(),
-        }
-    }
-
     pub fn as_data_column(&self) -> &DataColumnSidecar<T::EthSpec> {
         self.data_column.as_data_column()
     }
@@ -741,7 +734,7 @@ pub fn observe_gossip_data_column<T: BeaconChainTypes>(
     chain: &BeaconChain<T>,
 ) -> Result<(), GossipDataColumnError> {
     // Now the signature is valid, store the proposal so we don't accept another data column sidecar
-    // with the same `DataColumnIdentifier`.  It's important to double-check that the proposer still
+    // with the same `ColumnIndex`.  It's important to double-check that the proposer still
     // hasn't been observed so we don't have a race-condition when verifying two blocks
     // simultaneously.
     //
diff --git a/beacon_node/lighthouse_network/src/rpc/codec.rs b/beacon_node/lighthouse_network/src/rpc/codec.rs
index 2612172e61..f24074118e 100644
--- a/beacon_node/lighthouse_network/src/rpc/codec.rs
+++ b/beacon_node/lighthouse_network/src/rpc/codec.rs
@@ -16,11 +16,12 @@ use std::marker::PhantomData;
 use std::sync::Arc;
 use tokio_util::codec::{Decoder, Encoder};
 use types::{
-    BlobSidecar, ChainSpec, DataColumnSidecar, EthSpec, ForkContext, ForkName, Hash256,
-    LightClientBootstrap, LightClientFinalityUpdate, LightClientOptimisticUpdate,
-    LightClientUpdate, RuntimeVariableList, SignedBeaconBlock, SignedBeaconBlockAltair,
-    SignedBeaconBlockBase, SignedBeaconBlockBellatrix, SignedBeaconBlockCapella,
-    SignedBeaconBlockDeneb, SignedBeaconBlockElectra, SignedBeaconBlockFulu,
+    BlobSidecar, ChainSpec, DataColumnSidecar, DataColumnsByRootIdentifier, EthSpec, ForkContext,
+    ForkName, Hash256, LightClientBootstrap, LightClientFinalityUpdate,
+    LightClientOptimisticUpdate, LightClientUpdate, RuntimeVariableList, SignedBeaconBlock,
+    SignedBeaconBlockAltair, SignedBeaconBlockBase, SignedBeaconBlockBellatrix,
+    SignedBeaconBlockCapella, SignedBeaconBlockDeneb, SignedBeaconBlockElectra,
+    SignedBeaconBlockFulu,
 };
 use unsigned_varint::codec::Uvi;
 
@@ -596,10 +597,12 @@ fn handle_rpc_request<E: EthSpec>(
         ))),
         SupportedProtocol::DataColumnsByRootV1 => Ok(Some(RequestType::DataColumnsByRoot(
             DataColumnsByRootRequest {
-                data_column_ids: RuntimeVariableList::from_ssz_bytes(
-                    decoded_buffer,
-                    spec.max_request_data_column_sidecars as usize,
-                )?,
+                data_column_ids:
+                    <RuntimeVariableList<DataColumnsByRootIdentifier>>::from_ssz_bytes_with_nested(
+                        decoded_buffer,
+                        spec.max_request_blocks(current_fork),
+                        spec.number_of_columns as usize,
+                    )?,
             },
         ))),
         SupportedProtocol::PingV1 => Ok(Some(RequestType::Ping(Ping {
@@ -935,8 +938,8 @@ mod tests {
     use crate::types::{EnrAttestationBitfield, EnrSyncCommitteeBitfield};
     use types::{
         blob_sidecar::BlobIdentifier, data_column_sidecar::Cell, BeaconBlock, BeaconBlockAltair,
-        BeaconBlockBase, BeaconBlockBellatrix, BeaconBlockHeader, DataColumnIdentifier, EmptyBlock,
-        Epoch, FixedBytesExtended, FullPayload, KzgCommitment, KzgProof, Signature,
+        BeaconBlockBase, BeaconBlockBellatrix, BeaconBlockHeader, DataColumnsByRootIdentifier,
+        EmptyBlock, Epoch, FixedBytesExtended, FullPayload, KzgCommitment, KzgProof, Signature,
         SignedBeaconBlockHeader, Slot,
     };
 
@@ -1066,14 +1069,15 @@ mod tests {
         }
     }
 
-    fn dcbroot_request(spec: &ChainSpec) -> DataColumnsByRootRequest {
+    fn dcbroot_request(spec: &ChainSpec, fork_name: ForkName) -> DataColumnsByRootRequest {
+        let number_of_columns = spec.number_of_columns as usize;
         DataColumnsByRootRequest {
             data_column_ids: RuntimeVariableList::new(
-                vec![DataColumnIdentifier {
+                vec![DataColumnsByRootIdentifier {
                     block_root: Hash256::zero(),
-                    index: 0,
+                    columns: RuntimeVariableList::from_vec(vec![0, 1, 2], number_of_columns),
                 }],
-                spec.max_request_data_column_sidecars as usize,
+                spec.max_request_blocks(fork_name),
             )
             .unwrap(),
         }
@@ -1904,7 +1908,6 @@ mod tests {
             RequestType::MetaData(MetadataRequest::new_v1()),
             RequestType::BlobsByRange(blbrange_request()),
             RequestType::DataColumnsByRange(dcbrange_request()),
-            RequestType::DataColumnsByRoot(dcbroot_request(&chain_spec)),
             RequestType::MetaData(MetadataRequest::new_v2()),
         ];
         for req in requests.iter() {
@@ -1920,6 +1923,7 @@ mod tests {
                 RequestType::BlobsByRoot(blbroot_request(fork_name)),
                 RequestType::BlocksByRoot(bbroot_request_v1(fork_name)),
                 RequestType::BlocksByRoot(bbroot_request_v2(fork_name)),
+                RequestType::DataColumnsByRoot(dcbroot_request(&chain_spec, fork_name)),
             ]
         };
         for fork_name in ForkName::list_all() {
diff --git a/beacon_node/lighthouse_network/src/rpc/methods.rs b/beacon_node/lighthouse_network/src/rpc/methods.rs
index e6939e36d8..9fe2fef9e8 100644
--- a/beacon_node/lighthouse_network/src/rpc/methods.rs
+++ b/beacon_node/lighthouse_network/src/rpc/methods.rs
@@ -6,7 +6,6 @@ use serde::Serialize;
 use ssz::Encode;
 use ssz_derive::{Decode, Encode};
 use ssz_types::{typenum::U256, VariableList};
-use std::collections::BTreeMap;
 use std::fmt::Display;
 use std::marker::PhantomData;
 use std::ops::Deref;
@@ -16,9 +15,10 @@ use superstruct::superstruct;
 use types::blob_sidecar::BlobIdentifier;
 use types::light_client_update::MAX_REQUEST_LIGHT_CLIENT_UPDATES;
 use types::{
-    blob_sidecar::BlobSidecar, ChainSpec, ColumnIndex, DataColumnIdentifier, DataColumnSidecar,
-    Epoch, EthSpec, Hash256, LightClientBootstrap, LightClientFinalityUpdate,
-    LightClientOptimisticUpdate, LightClientUpdate, RuntimeVariableList, SignedBeaconBlock, Slot,
+    blob_sidecar::BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar,
+    DataColumnsByRootIdentifier, Epoch, EthSpec, Hash256, LightClientBootstrap,
+    LightClientFinalityUpdate, LightClientOptimisticUpdate, LightClientUpdate, RuntimeVariableList,
+    SignedBeaconBlock, Slot,
 };
 use types::{ForkContext, ForkName};
 
@@ -479,31 +479,20 @@ impl BlobsByRootRequest {
 #[derive(Clone, Debug, PartialEq)]
 pub struct DataColumnsByRootRequest {
     /// The list of beacon block roots and column indices being requested.
-    pub data_column_ids: RuntimeVariableList<DataColumnIdentifier>,
+    pub data_column_ids: RuntimeVariableList<DataColumnsByRootIdentifier>,
 }
 
 impl DataColumnsByRootRequest {
-    pub fn new(data_column_ids: Vec<DataColumnIdentifier>, spec: &ChainSpec) -> Self {
-        let data_column_ids = RuntimeVariableList::from_vec(
-            data_column_ids,
-            spec.max_request_data_column_sidecars as usize,
-        );
+    pub fn new(
+        data_column_ids: Vec<DataColumnsByRootIdentifier>,
+        max_request_blocks: usize,
+    ) -> Self {
+        let data_column_ids = RuntimeVariableList::from_vec(data_column_ids, max_request_blocks);
         Self { data_column_ids }
     }
 
-    pub fn new_single(block_root: Hash256, index: ColumnIndex, spec: &ChainSpec) -> Self {
-        Self::new(vec![DataColumnIdentifier { block_root, index }], spec)
-    }
-
-    pub fn group_by_ordered_block_root(&self) -> Vec<(Hash256, Vec<ColumnIndex>)> {
-        let mut column_indexes_by_block = BTreeMap::<Hash256, Vec<ColumnIndex>>::new();
-        for request_id in self.data_column_ids.as_slice() {
-            column_indexes_by_block
-                .entry(request_id.block_root)
-                .or_default()
-                .push(request_id.index);
-        }
-        column_indexes_by_block.into_iter().collect()
+    pub fn max_requested(&self) -> usize {
+        self.data_column_ids.iter().map(|id| id.columns.len()).sum()
     }
 }
 
diff --git a/beacon_node/lighthouse_network/src/rpc/protocol.rs b/beacon_node/lighthouse_network/src/rpc/protocol.rs
index 8fc1e9a5f4..820f50ac93 100644
--- a/beacon_node/lighthouse_network/src/rpc/protocol.rs
+++ b/beacon_node/lighthouse_network/src/rpc/protocol.rs
@@ -740,7 +740,7 @@ impl<E: EthSpec> RequestType<E> {
             RequestType::BlocksByRoot(req) => req.block_roots().len() as u64,
             RequestType::BlobsByRange(req) => req.max_blobs_requested(current_fork, spec),
             RequestType::BlobsByRoot(req) => req.blob_ids.len() as u64,
-            RequestType::DataColumnsByRoot(req) => req.data_column_ids.len() as u64,
+            RequestType::DataColumnsByRoot(req) => req.max_requested() as u64,
             RequestType::DataColumnsByRange(req) => req.max_requested::<E>(),
             RequestType::Ping(_) => 1,
             RequestType::MetaData(_) => 1,
diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs
index cf0e98cda8..2995a4d7e8 100644
--- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs
+++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs
@@ -1130,7 +1130,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         let processing_start_time = Instant::now();
         let block_root = verified_data_column.block_root();
         let data_column_slot = verified_data_column.slot();
-        let data_column_index = verified_data_column.id().index;
+        let data_column_index = verified_data_column.index();
 
         let result = self
             .chain
diff --git a/beacon_node/network/src/network_beacon_processor/rpc_methods.rs b/beacon_node/network/src/network_beacon_processor/rpc_methods.rs
index bc97f88492..96d5bc8181 100644
--- a/beacon_node/network/src/network_beacon_processor/rpc_methods.rs
+++ b/beacon_node/network/src/network_beacon_processor/rpc_methods.rs
@@ -360,24 +360,25 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
     ) -> Result<(), (RpcErrorResponse, &'static str)> {
         let mut send_data_column_count = 0;
 
-        for data_column_id in request.data_column_ids.as_slice() {
-            match self.chain.get_data_column_checking_all_caches(
-                data_column_id.block_root,
-                data_column_id.index,
+        for data_column_ids_by_root in request.data_column_ids.as_slice() {
+            match self.chain.get_data_columns_checking_all_caches(
+                data_column_ids_by_root.block_root,
+                data_column_ids_by_root.columns.as_slice(),
             ) {
-                Ok(Some(data_column)) => {
-                    send_data_column_count += 1;
-                    self.send_response(
-                        peer_id,
-                        inbound_request_id,
-                        Response::DataColumnsByRoot(Some(data_column)),
-                    );
+                Ok(data_columns) => {
+                    send_data_column_count += data_columns.len();
+                    for data_column in data_columns {
+                        self.send_response(
+                            peer_id,
+                            inbound_request_id,
+                            Response::DataColumnsByRoot(Some(data_column)),
+                        );
+                    }
                 }
-                Ok(None) => {} // no-op
                 Err(e) => {
                     // TODO(das): lower log level when feature is stabilized
                     error!(
-                        block_root = ?data_column_id.block_root,
+                        block_root = ?data_column_ids_by_root.block_root,
                         %peer_id,
                         error = ?e,
                         "Error getting data column"
@@ -389,7 +390,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
         debug!(
             %peer_id,
-            request = ?request.group_by_ordered_block_root(),
+            request = ?request.data_column_ids,
             returned = send_data_column_count,
             "Received DataColumnsByRoot Request"
         );
diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs
index d9eda651e7..58641f8606 100644
--- a/beacon_node/network/src/sync/network_context.rs
+++ b/beacon_node/network/src/sync/network_context.rs
@@ -875,7 +875,11 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
 
         self.send_network_msg(NetworkMessage::SendRequest {
             peer_id,
-            request: RequestType::DataColumnsByRoot(request.clone().into_request(&self.chain.spec)),
+            request: RequestType::DataColumnsByRoot(
+                request
+                    .clone()
+                    .try_into_request(self.fork_context.current_fork(), &self.chain.spec)?,
+            ),
             app_request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRoot(id)),
         })?;
 
diff --git a/beacon_node/network/src/sync/network_context/requests/data_columns_by_root.rs b/beacon_node/network/src/sync/network_context/requests/data_columns_by_root.rs
index 4e02737f08..09d7f4b3b7 100644
--- a/beacon_node/network/src/sync/network_context/requests/data_columns_by_root.rs
+++ b/beacon_node/network/src/sync/network_context/requests/data_columns_by_root.rs
@@ -1,6 +1,9 @@
 use lighthouse_network::rpc::methods::DataColumnsByRootRequest;
 use std::sync::Arc;
-use types::{ChainSpec, DataColumnIdentifier, DataColumnSidecar, EthSpec, Hash256};
+use types::{
+    ChainSpec, DataColumnSidecar, DataColumnsByRootIdentifier, EthSpec, ForkName, Hash256,
+    RuntimeVariableList,
+};
 
 use super::{ActiveRequestItems, LookupVerifyError};
 
@@ -11,17 +14,21 @@ pub struct DataColumnsByRootSingleBlockRequest {
 }
 
 impl DataColumnsByRootSingleBlockRequest {
-    pub fn into_request(self, spec: &ChainSpec) -> DataColumnsByRootRequest {
-        DataColumnsByRootRequest::new(
-            self.indices
-                .into_iter()
-                .map(|index| DataColumnIdentifier {
-                    block_root: self.block_root,
-                    index,
-                })
-                .collect(),
-            spec,
-        )
+    pub fn try_into_request(
+        self,
+        fork_name: ForkName,
+        spec: &ChainSpec,
+    ) -> Result<DataColumnsByRootRequest, &'static str> {
+        let number_of_columns = spec.number_of_columns as usize;
+        let columns = RuntimeVariableList::new(self.indices, number_of_columns)
+            .map_err(|_| "Number of indices exceeds total number of columns")?;
+        Ok(DataColumnsByRootRequest::new(
+            vec![DataColumnsByRootIdentifier {
+                block_root: self.block_root,
+                columns,
+            }],
+            spec.max_request_blocks(fork_name),
+        ))
     }
 }
 
diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs
index 565d7bc9f8..5863091cf0 100644
--- a/beacon_node/network/src/sync/tests/lookups.rs
+++ b/beacon_node/network/src/sync/tests/lookups.rs
@@ -979,18 +979,13 @@ impl TestRig {
                         request: RequestType::DataColumnsByRoot(request),
                         app_request_id:
                             AppRequestId::Sync(id @ SyncRequestId::DataColumnsByRoot { .. }),
-                    } if request
-                        .data_column_ids
-                        .to_vec()
-                        .iter()
-                        .any(|r| r.block_root == block_root) =>
-                    {
-                        let indices = request
+                    } => {
+                        let matching = request
                             .data_column_ids
-                            .to_vec()
                             .iter()
-                            .map(|cid| cid.index)
-                            .collect::<Vec<_>>();
+                            .find(|id| id.block_root == block_root)?;
+
+                        let indices = matching.columns.iter().copied().collect();
                         Some((*id, indices))
                     }
                     _ => None,
diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs
index 362c5d8014..d4b68357b2 100644
--- a/beacon_node/store/src/hot_cold_store.rs
+++ b/beacon_node/store/src/hot_cold_store.rs
@@ -117,19 +117,16 @@ impl<E: EthSpec> BlockCache<E> {
     pub fn get_blobs<'a>(&'a mut self, block_root: &Hash256) -> Option<&'a BlobSidecarList<E>> {
         self.blob_cache.get(block_root)
     }
-    pub fn get_data_columns(&mut self, block_root: &Hash256) -> Option<DataColumnSidecarList<E>> {
-        self.data_column_cache
-            .get(block_root)
-            .map(|map| map.values().cloned().collect::<Vec<_>>())
-    }
-    pub fn get_data_column<'a>(
-        &'a mut self,
+    // Note: data columns are all individually cached, hence there's no guarantee that
+    // `data_column_cache.get(block_root)` will return all custody columns.
+    pub fn get_data_column(
+        &mut self,
         block_root: &Hash256,
         column_index: &ColumnIndex,
-    ) -> Option<&'a Arc<DataColumnSidecar<E>>> {
+    ) -> Option<Arc<DataColumnSidecar<E>>> {
         self.data_column_cache
             .get(block_root)
-            .and_then(|map| map.get(column_index))
+            .and_then(|map| map.get(column_index).cloned())
     }
     pub fn delete_block(&mut self, block_root: &Hash256) {
         let _ = self.block_cache.pop(block_root);
@@ -2031,38 +2028,19 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             })
     }
 
-    /// Fetch columns for a given block from the store.
+    /// Fetch all columns for a given block from the store.
     pub fn get_data_columns(
         &self,
         block_root: &Hash256,
     ) -> Result<Option<DataColumnSidecarList<E>>, Error> {
-        if let Some(columns) = self.block_cache.lock().get_data_columns(block_root) {
-            metrics::inc_counter(&metrics::BEACON_DATA_COLUMNS_CACHE_HIT_COUNT);
-            return Ok(Some(columns));
-        }
+        let column_indices = self.get_data_column_keys(*block_root)?;
 
-        let columns = self
-            .blobs_db
-            .iter_column_from::<Vec<u8>>(DBColumn::BeaconDataColumn, block_root.as_slice())
-            .take_while(|res| {
-                res.as_ref()
-                    .is_ok_and(|(key, _)| key.starts_with(block_root.as_slice()))
-            })
-            .map(|result| {
-                let (_key, value) = result?;
-                let column = DataColumnSidecar::<E>::from_ssz_bytes(&value).map(Arc::new)?;
-                self.block_cache
-                    .lock()
-                    .put_data_column(*block_root, column.clone());
-                Ok(column)
-            })
-            .collect::<Result<DataColumnSidecarList<E>, Error>>()?;
+        let columns: DataColumnSidecarList<E> = column_indices
+            .into_iter()
+            .filter_map(|col_index| self.get_data_column(block_root, &col_index).transpose())
+            .collect::<Result<_, _>>()?;
 
-        if columns.is_empty() {
-            Ok(None)
-        } else {
-            Ok(Some(columns))
-        }
+        Ok((!columns.is_empty()).then_some(columns))
     }
 
     /// Fetch blobs for a given block from the store.
@@ -2127,7 +2105,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             .get_data_column(block_root, column_index)
         {
             metrics::inc_counter(&metrics::BEACON_DATA_COLUMNS_CACHE_HIT_COUNT);
-            return Ok(Some(data_column.clone()));
+            return Ok(Some(data_column));
         }
 
         match self.blobs_db.get_bytes(
diff --git a/consensus/types/src/chain_spec.rs b/consensus/types/src/chain_spec.rs
index 2b29ef1f10..7b9950db91 100644
--- a/consensus/types/src/chain_spec.rs
+++ b/consensus/types/src/chain_spec.rs
@@ -1,6 +1,6 @@
 use crate::application_domain::{ApplicationDomain, APPLICATION_DOMAIN_BUILDER};
 use crate::blob_sidecar::BlobIdentifier;
-use crate::data_column_sidecar::DataColumnIdentifier;
+use crate::data_column_sidecar::DataColumnsByRootIdentifier;
 use crate::*;
 use int_to_bytes::int_to_bytes4;
 use safe_arith::{ArithError, SafeArith};
@@ -1754,15 +1754,21 @@ fn max_blobs_by_root_request_common(max_request_blob_sidecars: u64) -> usize {
     .len()
 }
 
-fn max_data_columns_by_root_request_common(max_request_data_column_sidecars: u64) -> usize {
-    let max_request_data_column_sidecars = max_request_data_column_sidecars as usize;
-    let empty_data_column_id = DataColumnIdentifier {
+fn max_data_columns_by_root_request_common(
+    max_request_blocks: u64,
+    number_of_columns: u64,
+) -> usize {
+    let max_request_blocks = max_request_blocks as usize;
+    let number_of_columns = number_of_columns as usize;
+
+    let empty_data_columns_by_root_id = DataColumnsByRootIdentifier {
         block_root: Hash256::zero(),
-        index: 0,
+        columns: RuntimeVariableList::from_vec(vec![0; number_of_columns], number_of_columns),
     };
-    RuntimeVariableList::from_vec(
-        vec![empty_data_column_id; max_request_data_column_sidecars],
-        max_request_data_column_sidecars,
+
+    RuntimeVariableList::<DataColumnsByRootIdentifier>::from_vec(
+        vec![empty_data_columns_by_root_id; max_request_blocks],
+        max_request_blocks,
     )
     .as_ssz_bytes()
     .len()
@@ -1781,7 +1787,10 @@ fn default_max_blobs_by_root_request() -> usize {
 }
 
 fn default_data_columns_by_root_request() -> usize {
-    max_data_columns_by_root_request_common(default_max_request_data_column_sidecars())
+    max_data_columns_by_root_request_common(
+        default_max_request_blocks_deneb(),
+        default_number_of_columns(),
+    )
 }
 
 impl Default for Config {
@@ -2082,7 +2091,8 @@ impl Config {
             ),
             max_blobs_by_root_request: max_blobs_by_root_request_common(max_request_blob_sidecars),
             max_data_columns_by_root_request: max_data_columns_by_root_request_common(
-                max_request_data_column_sidecars,
+                max_request_blocks_deneb,
+                number_of_columns,
             ),
 
             number_of_columns,
diff --git a/consensus/types/src/data_column_sidecar.rs b/consensus/types/src/data_column_sidecar.rs
index 03ab6a74f8..d2802670b6 100644
--- a/consensus/types/src/data_column_sidecar.rs
+++ b/consensus/types/src/data_column_sidecar.rs
@@ -1,7 +1,9 @@
 use crate::beacon_block_body::{KzgCommitments, BLOB_KZG_COMMITMENTS_INDEX};
 use crate::test_utils::TestRandom;
-use crate::BeaconStateError;
-use crate::{BeaconBlockHeader, Epoch, EthSpec, Hash256, SignedBeaconBlockHeader, Slot};
+use crate::{
+    BeaconBlockHeader, BeaconStateError, Epoch, EthSpec, Hash256, RuntimeVariableList,
+    SignedBeaconBlockHeader, Slot,
+};
 use bls::Signature;
 use derivative::Derivative;
 use kzg::Error as KzgError;
@@ -9,11 +11,10 @@ use kzg::{KzgCommitment, KzgProof};
 use merkle_proof::verify_merkle_proof;
 use safe_arith::ArithError;
 use serde::{Deserialize, Serialize};
-use ssz::Encode;
+use ssz::{DecodeError, Encode};
 use ssz_derive::{Decode, Encode};
 use ssz_types::Error as SszError;
 use ssz_types::{FixedVector, VariableList};
-use std::hash::Hash;
 use std::sync::Arc;
 use test_random_derive::TestRandom;
 use tree_hash::TreeHash;
@@ -23,13 +24,47 @@ pub type ColumnIndex = u64;
 pub type Cell<E> = FixedVector<u8, <E as EthSpec>::BytesPerCell>;
 pub type DataColumn<E> = VariableList<Cell<E>, <E as EthSpec>::MaxBlobCommitmentsPerBlock>;
 
-/// Container of the data that identifies an individual data column.
-#[derive(
-    Serialize, Deserialize, Encode, Decode, TreeHash, Copy, Clone, Debug, PartialEq, Eq, Hash,
-)]
-pub struct DataColumnIdentifier {
+/// Identifies a set of data columns associated with a specific beacon block.
+#[derive(Encode, Clone, Debug, PartialEq)]
+pub struct DataColumnsByRootIdentifier {
     pub block_root: Hash256,
-    pub index: ColumnIndex,
+    pub columns: RuntimeVariableList<ColumnIndex>,
+}
+
+impl RuntimeVariableList<DataColumnsByRootIdentifier> {
+    pub fn from_ssz_bytes_with_nested(
+        bytes: &[u8],
+        max_len: usize,
+        num_columns: usize,
+    ) -> Result<Self, DecodeError> {
+        if bytes.is_empty() {
+            return Ok(RuntimeVariableList::empty(max_len));
+        }
+
+        let vec = ssz::decode_list_of_variable_length_items::<Vec<u8>, Vec<Vec<u8>>>(
+            bytes,
+            Some(max_len),
+        )?
+        .into_iter()
+        .map(|bytes| {
+            let mut builder = ssz::SszDecoderBuilder::new(&bytes);
+            builder.register_type::<Hash256>()?;
+            builder.register_anonymous_variable_length_item()?;
+
+            let mut decoder = builder.build()?;
+            let block_root = decoder.decode_next()?;
+            let columns = decoder.decode_next_with(|bytes| {
+                RuntimeVariableList::from_ssz_bytes(bytes, num_columns)
+            })?;
+            Ok(DataColumnsByRootIdentifier {
+                block_root,
+                columns,
+            })
+        })
+        .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(RuntimeVariableList::from_vec(vec, max_len))
+    }
 }
 
 pub type DataColumnSidecarList<E> = Vec<Arc<DataColumnSidecar<E>>>;
@@ -132,13 +167,6 @@ impl<E: EthSpec> DataColumnSidecar<E> {
         .as_ssz_bytes()
         .len()
     }
-
-    pub fn id(&self) -> DataColumnIdentifier {
-        DataColumnIdentifier {
-            block_root: self.block_root(),
-            index: self.index,
-        }
-    }
 }
 
 #[derive(Debug)]
@@ -178,3 +206,45 @@ impl From<SszError> for DataColumnSidecarError {
         Self::SszError(e)
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use bls::FixedBytesExtended;
+
+    #[test]
+    fn round_trip_dcbroot_list() {
+        let max_outer = 5;
+        let max_inner = 10;
+
+        let data = vec![
+            DataColumnsByRootIdentifier {
+                block_root: Hash256::from_low_u64_be(10),
+                columns: RuntimeVariableList::<ColumnIndex>::from_vec(vec![1u64, 2, 3], max_inner),
+            },
+            DataColumnsByRootIdentifier {
+                block_root: Hash256::from_low_u64_be(20),
+                columns: RuntimeVariableList::<ColumnIndex>::from_vec(vec![4u64, 5], max_inner),
+            },
+        ];
+
+        let list = RuntimeVariableList::from_vec(data.clone(), max_outer);
+
+        let ssz_bytes = list.as_ssz_bytes();
+
+        let decoded =
+            RuntimeVariableList::<DataColumnsByRootIdentifier>::from_ssz_bytes_with_nested(
+                &ssz_bytes, max_outer, max_inner,
+            )
+            .expect("should decode list of DataColumnsByRootIdentifier");
+
+        assert_eq!(decoded.len(), data.len());
+        for (original, decoded) in data.iter().zip(decoded.iter()) {
+            assert_eq!(decoded.block_root, original.block_root);
+            assert_eq!(
+                decoded.columns.iter().copied().collect::<Vec<_>>(),
+                original.columns.iter().copied().collect::<Vec<_>>()
+            );
+        }
+    }
+}
diff --git a/consensus/types/src/lib.rs b/consensus/types/src/lib.rs
index 1d39c89cab..70f07f0109 100644
--- a/consensus/types/src/lib.rs
+++ b/consensus/types/src/lib.rs
@@ -149,7 +149,7 @@ pub use crate::config_and_preset::{
 pub use crate::consolidation_request::ConsolidationRequest;
 pub use crate::contribution_and_proof::ContributionAndProof;
 pub use crate::data_column_sidecar::{
-    ColumnIndex, DataColumnIdentifier, DataColumnSidecar, DataColumnSidecarList,
+    ColumnIndex, DataColumnSidecar, DataColumnSidecarList, DataColumnsByRootIdentifier,
 };
 pub use crate::data_column_subnet_id::DataColumnSubnetId;
 pub use crate::deposit::{Deposit, DEPOSIT_TREE_DEPTH};
diff --git a/scripts/local_testnet/network_params_das.yaml b/scripts/local_testnet/network_params_das.yaml
index d47dfa6b5a..628b2696a5 100644
--- a/scripts/local_testnet/network_params_das.yaml
+++ b/scripts/local_testnet/network_params_das.yaml
@@ -1,39 +1,37 @@
 participants:
   - cl_type: lighthouse
     cl_image: lighthouse:local
-    el_image: ethpandaops/geth:engine-getblobs-v2-3676b56
+    el_image: ethpandaops/geth:marius-engine-getblobs-v2
     cl_extra_params:
       - --subscribe-all-data-column-subnets
       - --subscribe-all-subnets
-      # Note: useful for testing range sync (only produce block if node is in sync to prevent forking)
+      # Note: useful for testing range sync (only produce block if the node is in sync to prevent forking)
       - --sync-tolerance-epochs=0
       - --target-peers=3
     count: 2
   - cl_type: lighthouse
     cl_image: lighthouse:local
-    el_image: ethpandaops/geth:engine-getblobs-v2-3676b56
+    el_image: ethpandaops/geth:marius-engine-getblobs-v2
     cl_extra_params:
-      # Note: useful for testing range sync (only produce block if node is in sync to prevent forking)
+      # Note: useful for testing range sync (only produce block if the node is in sync to prevent forking)
       - --sync-tolerance-epochs=0
       - --target-peers=3
     count: 2
 network_params:
-  electra_fork_epoch: 1
-  fulu_fork_epoch: 2
+  electra_fork_epoch: 0
+  fulu_fork_epoch: 1
   seconds_per_slot: 6
-  max_blobs_per_block_electra: 64
-  target_blobs_per_block_electra: 48
-  max_blobs_per_block_fulu: 64
-  target_blobs_per_block_fulu: 48
 snooper_enabled: false
 global_log_level: debug
 additional_services:
   - dora
-  - spamoor_blob
+  - spamoor
   - prometheus_grafana
-dora_params:
-  image: ethpandaops/dora:fulu-support
-spamoor_blob_params:
-  # Throughput of spamoor
-  # Defaults to 3
-  throughput: 32
\ No newline at end of file
+spamoor_params:
+  spammers:
+    - scenario: eoatx
+      config:
+        throughput: 200
+    - scenario: blobs
+      config:
+        throughput: 20
\ No newline at end of file
diff --git a/testing/ef_tests/src/type_name.rs b/testing/ef_tests/src/type_name.rs
index dfee385958..387e77310d 100644
--- a/testing/ef_tests/src/type_name.rs
+++ b/testing/ef_tests/src/type_name.rs
@@ -58,7 +58,7 @@ type_name_generic!(BeaconBlockBodyFulu, "BeaconBlockBody");
 type_name!(BeaconBlockHeader);
 type_name_generic!(BeaconState);
 type_name!(BlobIdentifier);
-type_name!(DataColumnIdentifier);
+type_name!(DataColumnsByRootIdentifier);
 type_name_generic!(BlobSidecar);
 type_name_generic!(DataColumnSidecar);
 type_name!(Checkpoint);
diff --git a/testing/ef_tests/tests/tests.rs b/testing/ef_tests/tests/tests.rs
index 3948708edf..d333cdbb11 100644
--- a/testing/ef_tests/tests/tests.rs
+++ b/testing/ef_tests/tests/tests.rs
@@ -667,11 +667,13 @@ mod ssz_static {
     }
 
     #[test]
-    fn data_column_identifier() {
-        SszStaticHandler::<DataColumnIdentifier, MinimalEthSpec>::default()
-            .run_for_feature(FeatureName::Fulu);
-        SszStaticHandler::<DataColumnIdentifier, MainnetEthSpec>::default()
-            .run_for_feature(FeatureName::Fulu);
+    #[ignore]
+    // TODO(das): enable once EF tests are updated to latest release.
+    fn data_column_by_root_identifier() {
+        // SszStaticHandler::<DataColumnsByRootIdentifier, MinimalEthSpec>::default()
+        //     .run_for_feature(FeatureName::Fulu);
+        // SszStaticHandler::<DataColumnsByRootIdentifier, MainnetEthSpec>::default()
+        //     .run_for_feature(FeatureName::Fulu);
     }
 
     #[test]