Address comments

Author: kyli87

kernel/src/engine/plans/json.rs

@@ -7,8 +7,8 @@ use url::Url;
 use crate::plans::{Plan, PlanExecutor, QueryPlanBuilder};
 use crate::schema::SchemaRef;
 use crate::{
-    DeltaResult, EngineData, FileDataReadResultIterator, FileMeta, FilteredEngineData, JsonHandler,
-    PredicateRef,
+    DeltaResult, EngineData, Error, FileDataReadResultIterator, FileMeta, FilteredEngineData,
+    JsonHandler, PredicateRef,
 };
 
 /// A [`JsonHandler`] that delegates to a [`PlanExecutor`].
@@ -30,7 +30,9 @@ impl JsonHandler for PlanBasedJsonHandler {
         _json_strings: Box<dyn EngineData>,
         _output_schema: SchemaRef,
     ) -> DeltaResult<Box<dyn EngineData>> {
-        todo!("PlanBasedJsonHandler does not support parse_json yet");
+        Err(Error::unsupported(
+            "PlanBasedJsonHandler does not support parse_json yet",
+        ))
     }
 
     fn read_json_files(
@@ -52,7 +54,9 @@ impl JsonHandler for PlanBasedJsonHandler {
         _data: Box<dyn Iterator<Item = DeltaResult<FilteredEngineData>> + Send + '_>,
         _overwrite: bool,
     ) -> DeltaResult<()> {
-        todo!("PlanBasedJsonHandler does not support write_json_file yet");
+        Err(Error::unsupported(
+            "PlanBasedJsonHandler does not support write_json_file yet",
+        ))
     }
 }
 

kernel/src/engine/plans/parquet.rs

@@ -7,8 +7,8 @@ use url::Url;
 use crate::plans::{Plan, PlanExecutor, QueryPlanBuilder};
 use crate::schema::SchemaRef;
 use crate::{
-    DeltaResult, EngineData, FileDataReadResultIterator, FileMeta, ParquetFooter, ParquetHandler,
-    PredicateRef,
+    DeltaResult, EngineData, Error, FileDataReadResultIterator, FileMeta, ParquetFooter,
+    ParquetHandler, PredicateRef,
 };
 
 /// A [`ParquetHandler`] that delegates to a [`PlanExecutor`].
@@ -43,11 +43,15 @@ impl ParquetHandler for PlanBasedParquetHandler {
         _location: Url,
         _data: Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>> + Send>,
     ) -> DeltaResult<()> {
-        todo!("PlanBasedParquetHandler does not support write_parquet_file");
+        Err(Error::unsupported(
+            "PlanBasedParquetHandler does not support write_parquet_file yet",
+        ))
     }
 
     fn read_parquet_footer(&self, _file: &FileMeta) -> DeltaResult<ParquetFooter> {
-        todo!("PlanBasedParquetHandler does not support read_parquet_footer");
+        Err(Error::unsupported(
+            "PlanBasedParquetHandler does not support read_parquet_footer yet",
+        ))
     }
 }
 

kernel/src/engine/plans/storage.rs

@@ -3,7 +3,7 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
-use itertools::Itertools;
+use itertools::Itertools as _;
 use url::Url;
 
 use crate::plans::{IoOperation, Plan, PlanExecutor, PlanResult};
@@ -29,20 +29,18 @@ impl StorageHandler for PlanBasedStorageHandler {
         &self,
         path: &Url,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<FileMeta>>>> {
-        let iter = self
+        Ok(self
             .execute_io(IoOperation::file_listing(path.clone()))?
-            .into_file_meta()?;
-        Ok(iter)
+            .into_file_meta()?)
     }
 
     fn read_files(
         &self,
         files: Vec<FileSlice>,
     ) -> DeltaResult<Box<dyn Iterator<Item = DeltaResult<Bytes>>>> {
-        let iter = self
+        Ok(self
             .execute_io(IoOperation::read_bytes(files))?
-            .into_bytes()?;
-        Ok(iter)
+            .into_bytes()?)
     }
 
     fn copy_atomic(&self, src: &Url, dest: &Url) -> DeltaResult<()> {