feat(flow-control): do flow control during evaluation too (#710)

Author: georgeh0

src/execution/dumper.rs

@@ -136,6 +136,10 @@ impl<'a> Dumper<'a> {
         key: value::KeyValue,
         file_path: PathBuf,
     ) -> Result<()> {
+        let _permit = import_op
+            .concurrency_controller
+            .acquire(concur_control::BYTES_UNKNOWN_YET)
+            .await?;
         let mut collected_values_buffer = Vec::new();
         let (exports, error) = match self
             .evaluate_source_entry(import_op_idx, import_op, &key, &mut collected_values_buffer)

src/execution/evaluator.rs

@@ -464,6 +464,11 @@ pub async fn evaluate_source_entry(
     source_value: value::FieldValues,
     memory: &EvaluationMemory,
 ) -> Result<EvaluateSourceEntryOutput> {
+    let _permit = src_eval_ctx
+        .import_op
+        .concurrency_controller
+        .acquire_bytes_with_reservation(|| source_value.estimated_byte_size())
+        .await?;
     let root_schema = &src_eval_ctx.schema.schema;
     let root_scope_value = ScopeValueBuilder::new(root_schema.fields.len());
     let root_scope_entry = ScopeEntry::new(

src/execution/source_indexer.rs

@@ -163,15 +163,6 @@ impl SourceIndexingContext {
 
             {
                 let _processing_permit = processing_sem.acquire().await?;
-                let _concur_permit = match &source_data.value {
-                    interface::SourceValue::Existence(value) => {
-                        import_op
-                            .concurrency_controller
-                            .acquire_bytes_with_reservation(|| value.estimated_byte_size())
-                            .await?
-                    }
-                    interface::SourceValue::NonExistence => None,
-                };
                 let result = row_indexer::update_source_row(
                     &SourceRowEvaluationContext {
                         plan: &plan,