clean up of python rewrite

daft/iceberg/iceberg_scan.py

@@ -87,6 +87,7 @@ def partitioning_keys(self) -> list[PartitionField]:
 
     def _iceberg_record_to_partition_spec(self, record: Record) -> daft.table.Table | None:
         arrays = dict()
+        assert len(record._position_to_field_name) == len(self._partition_keys)
         for name, value, pfield in zip(record._position_to_field_name, record.record_fields(), self._partition_keys):
             field = Field._from_pyfield(pfield.field)
             field_name = field.name
@@ -103,7 +104,7 @@ def to_scan_tasks(self, pushdowns: Pushdowns) -> Iterator[ScanTask]:
         limit = pushdowns.limit
         iceberg_tasks = self._table.scan(limit=limit).plan_files()
 
-        limit_files = limit is not None and pushdowns.filters is None
+        limit_files = limit is not None and pushdowns.filters is None and pushdowns.partition_filters is None
 
         scan_tasks = []
 

src/daft-scan/src/lib.rs

@@ -462,7 +462,7 @@ pub struct Pushdowns {
 
 impl Default for Pushdowns {
     fn default() -> Self {
-        Self::new(None, None,  None, None)
+        Self::new(None, None, None, None)
     }
 }
 

src/daft-scan/src/python.rs

@@ -214,18 +214,6 @@ partitioning_keys:\n",
         ) -> common_error::DaftResult<
             Box<dyn Iterator<Item = common_error::DaftResult<crate::ScanTaskRef>>>,
         > {
-            println!("{:?}", pushdowns.filters);
-            if let Some(pred) = pushdowns.filters.as_deref() {
-                let transformed = rewrite_predicate_for_partitioning(
-                    pred.as_ref().clone(),
-                    self.partitioning_keys.as_slice(),
-                )?;
-                println!("before {}", pred);
-                for t in transformed {
-                    println!(" {t}");
-                }
-            }
-
             let scan_tasks = Python::with_gil(|py| {
                 let pypd = PyPushdowns(pushdowns.into()).into_py(py);
                 let pyiter =