fix: check if eligible to read (delta-io#3771)

# Description
- closes delta-io#3765

Signed-off-by: Ion Koutsouris <15728914+ion-elgreco@users.noreply.github.com>

Author: ion-elgreco

crates/core/src/delta_datafusion/table_provider.rs

@@ -57,7 +57,7 @@ use crate::delta_datafusion::{
     LogDataHandler,
 };
 use crate::kernel::schema::cast::cast_record_batch;
-use crate::kernel::transaction::CommitBuilder;
+use crate::kernel::transaction::{CommitBuilder, PROTOCOL};
 use crate::kernel::{Action, Add, Remove};
 use crate::operations::write::writer::{DeltaWriter, WriterConfig};
 use crate::operations::write::WriterStatsConfig;
@@ -451,6 +451,7 @@ impl<'a> DeltaScanBuilder<'a> {
     }
 
     pub async fn build(self) -> DeltaResult<DeltaScan> {
+        PROTOCOL.can_read_from(self.snapshot)?;
         let config = match self.config {
             Some(config) => config,
             None => DeltaScanConfigBuilder::new().build(self.snapshot)?,

crates/core/src/operations/load_cdf.rs

@@ -32,6 +32,7 @@ use tracing::log;
 
 use crate::delta_datafusion::{register_store, DataFusionMixins};
 use crate::errors::DeltaResult;
+use crate::kernel::transaction::PROTOCOL;
 use crate::kernel::{Action, Add, AddCDCFile, CommitInfo};
 use crate::logstore::{get_actions, LogStoreRef};
 use crate::table::state::DeltaTableState;
@@ -328,6 +329,8 @@ impl CdfLoadBuilder {
         session_sate: &SessionState,
         filters: Option<&Arc<dyn PhysicalExpr>>,
     ) -> DeltaResult<Arc<dyn ExecutionPlan>> {
+        PROTOCOL.can_read_from(&self.snapshot)?;
+
         let (cdc, add, remove) = self.determine_files_to_read().await?;
         register_store(self.log_store.clone(), session_sate.runtime_env().clone());