feat: raise if files not loaded

crates/core/src/errors.rs

@@ -221,6 +221,9 @@ pub enum DeltaTableError {
     #[error("Table has not yet been initialized")]
     NotInitialized,
 
+    #[error("Table has not yet been initialized with files, therefore {0} is not supported")]
+    NotInitializedWithFiles(String),
+
     #[error("Change Data not enabled for version: {version}, Start: {start}, End: {end}")]
     ChangeDataNotRecorded { version: i64, start: i64, end: i64 },
 

crates/core/src/kernel/snapshot/mod.rs

@@ -193,6 +193,11 @@ impl Snapshot {
         &self.protocol
     }
 
+    /// Get the table config which is loaded with of the snapshot
+    pub fn load_config(&self) -> &DeltaTableConfig {
+        &self.config
+    }
+
     /// Get the table root of the snapshot
     pub fn table_root(&self) -> Path {
         Path::from(self.table_url.clone())
@@ -535,6 +540,11 @@ impl EagerSnapshot {
         self.snapshot.table_root()
     }
 
+    /// Get the table config which is loaded with of the snapshot
+    pub fn load_config(&self) -> &DeltaTableConfig {
+        &self.snapshot.load_config()
+    }
+
     /// Well known table configuration
     pub fn table_config(&self) -> TableConfig<'_> {
         self.snapshot.table_config()

crates/core/src/operations/constraints.rs

@@ -89,6 +89,12 @@ impl std::future::IntoFuture for ConstraintBuilder {
         let this = self;
 
         Box::pin(async move {
+            if !this.snapshot.load_config().require_files {
+                return Err(DeltaTableError::NotInitializedWithFiles(
+                    "ADD CONSTRAINTS".into(),
+                ));
+            }
+
             let name = match this.name {
                 Some(v) => v,
                 None => return Err(DeltaTableError::Generic("No name provided".to_string())),

crates/core/src/operations/delete.rs

@@ -293,6 +293,10 @@ async fn execute(
     writer_properties: Option<WriterProperties>,
     mut commit_properties: CommitProperties,
 ) -> DeltaResult<(DeltaTableState, DeleteMetrics)> {
+    if !&snapshot.load_config().require_files {
+        return Err(DeltaTableError::NotInitializedWithFiles("DELETE".into()));
+    }
+
     let exec_start = Instant::now();
     let mut metrics = DeleteMetrics::default();
 

crates/core/src/operations/load.rs

@@ -51,6 +51,9 @@ impl std::future::IntoFuture for LoadBuilder {
 
         Box::pin(async move {
             PROTOCOL.can_read_from(&this.snapshot.snapshot)?;
+            if !this.snapshot.load_config().require_files {
+                return Err(DeltaTableError::NotInitializedWithFiles("reading".into()));
+            }
 
             let table = DeltaTable::new_with_state(this.log_store, this.snapshot);
             let schema = table.snapshot()?.arrow_schema()?;

crates/core/src/operations/merge/mod.rs

@@ -690,6 +690,10 @@ async fn execute(
     not_match_target_operations: Vec<MergeOperationConfig>,
     not_match_source_operations: Vec<MergeOperationConfig>,
 ) -> DeltaResult<(DeltaTableState, MergeMetrics)> {
+    if !snapshot.load_config().require_files {
+        return Err(DeltaTableError::NotInitializedWithFiles("MERGE".into()));
+    }
+
     let mut metrics = MergeMetrics::default();
     let exec_start = Instant::now();
     // Determining whether we should write change data once so that computation of change data can

crates/core/src/operations/optimize.rs

@@ -296,6 +296,9 @@ impl<'a> std::future::IntoFuture for OptimizeBuilder<'a> {
 
         Box::pin(async move {
             PROTOCOL.can_write_to(&this.snapshot.snapshot)?;
+            if !&this.snapshot.load_config().require_files {
+                return Err(DeltaTableError::NotInitializedWithFiles("OPTIMIZE".into()));
+            }
 
             let writer_properties = this.writer_properties.unwrap_or_else(|| {
                 WriterProperties::builder()

crates/core/src/operations/update.rs

@@ -238,6 +238,9 @@ async fn execute(
     // For files that were identified, scan for records that match the predicate,
     // perform update operations, and then commit add and remove actions to
     // the log.
+    if !&snapshot.load_config().require_files {
+        return Err(DeltaTableError::NotInitializedWithFiles("UPDATE".into()));
+    }
 
     let update_planner = DeltaPlanner::<UpdateMetricExtensionPlanner> {
         extension_planner: UpdateMetricExtensionPlanner {},

crates/core/src/operations/vacuum.rs

@@ -240,8 +240,11 @@ impl std::future::IntoFuture for VacuumBuilder {
 
     fn into_future(self) -> Self::IntoFuture {
         let this = self;
-
         Box::pin(async move {
+            if !&this.snapshot.load_config().require_files {
+                return Err(DeltaTableError::NotInitializedWithFiles("VACUUM".into()));
+            }
+
             let plan = this.create_vacuum_plan().await?;
             if this.dry_run {
                 return Ok((

crates/core/src/operations/write.rs

@@ -789,6 +789,9 @@ impl std::future::IntoFuture for WriteBuilder {
             if this.mode == SaveMode::Overwrite {
                 if let Some(snapshot) = &this.snapshot {
                     PROTOCOL.check_append_only(&snapshot.snapshot)?;
+                    if !snapshot.load_config().require_files {
+                        return Err(DeltaTableError::NotInitializedWithFiles("WRITE".into()));
+                    }
                 }
             }
             if this.schema_mode == Some(SchemaMode::Overwrite) && this.mode != SaveMode::Overwrite {

crates/core/src/table/state.rs

@@ -181,6 +181,11 @@ impl DeltaTableState {
         self.snapshot.schema()
     }
 
+    /// Get the table config which is loaded with of the snapshot
+    pub fn load_config(&self) -> &DeltaTableConfig {
+        &self.snapshot.load_config()
+    }
+
     /// Well known table configuration
     pub fn table_config(&self) -> TableConfig<'_> {
         self.snapshot.table_config()

python/deltalake/_internal.pyi

@@ -45,6 +45,7 @@ class RawDeltaTable:
     ) -> bool: ...
     def table_uri(self) -> str: ...
     def version(self) -> int: ...
+    def has_files(self) -> bool: ...
     def get_add_file_sizes(self) -> Dict[str, int]: ...
     def get_latest_version(self) -> int: ...
     def get_num_index_cols(self) -> int: ...

python/deltalake/table.py

@@ -39,6 +39,7 @@
     import os
 
 from deltalake._internal import (
+    DeltaError,
     PyMergeBuilder,
     RawDeltaTable,
 )
@@ -1138,6 +1139,9 @@ def to_pyarrow_dataset(
         Returns:
             the PyArrow dataset in PyArrow
         """
+        if not self._table.has_files():
+            raise DeltaError("Table is instantiated without files.")
+
         table_protocol = self.protocol()
         if (
             table_protocol.min_reader_version > MAX_SUPPORTED_READER_VERSION
@@ -1158,7 +1162,6 @@ def to_pyarrow_dataset(
                 raise DeltaProtocolError(
                     f"The table has set these reader features: {missing_features} but these are not yet supported by the deltalake reader."
                 )
-
         if not filesystem:
             filesystem = pa_fs.PyFileSystem(
                 DeltaStorageHandler.from_table(

python/src/lib.rs

@@ -50,6 +50,7 @@ use deltalake::protocol::{DeltaOperation, SaveMode};
 use deltalake::storage::IORuntime;
 use deltalake::DeltaTableBuilder;
 use deltalake::{DeltaOps, DeltaResult};
+use error::DeltaError;
 use futures::future::join_all;
 
 use pyo3::exceptions::{PyRuntimeError, PyValueError};
@@ -166,6 +167,10 @@ impl RawDeltaTable {
         Ok(self._table.version())
     }
 
+    pub fn has_files(&self) -> PyResult<bool> {
+        Ok(self._table.config.require_files)
+    }
+
     pub fn metadata(&self) -> PyResult<RawDeltaTableMetaData> {
         let metadata = self._table.metadata().map_err(PythonError::from)?;
         Ok(RawDeltaTableMetaData {
@@ -273,6 +278,9 @@ impl RawDeltaTable {
         py: Python,
         partition_filters: Option<Vec<(PyBackedStr, PyBackedStr, PartitionFilterValue)>>,
     ) -> PyResult<Vec<String>> {
+        if !self.has_files()? {
+            return Err(DeltaError::new_err("Table is instantiated without files."));
+        }
         py.allow_threads(|| {
             if let Some(filters) = partition_filters {
                 let filters = convert_partition_filters(filters).map_err(PythonError::from)?;
@@ -298,6 +306,10 @@ impl RawDeltaTable {
         &self,
         partition_filters: Option<Vec<(PyBackedStr, PyBackedStr, PartitionFilterValue)>>,
     ) -> PyResult<Vec<String>> {
+        if !self._table.config.require_files {
+            return Err(DeltaError::new_err("Table is initiated without files."));
+        }
+
         if let Some(filters) = partition_filters {
             let filters = convert_partition_filters(filters).map_err(PythonError::from)?;
             Ok(self
@@ -1073,6 +1085,9 @@ impl RawDeltaTable {
     }
 
     pub fn get_add_actions(&self, flatten: bool) -> PyResult<PyArrowType<RecordBatch>> {
+        if !self.has_files()? {
+            return Err(DeltaError::new_err("Table is instantiated without files."));
+        }
         Ok(PyArrowType(
             self._table
                 .snapshot()