feat: add max_file_size support to AzureBlob source

Author: prabhath004

docs/docs/sources/azureblob.md

@@ -63,15 +63,19 @@ The spec takes the following fields:
 *   `excluded_patterns` (`list[str]`, optional): a list of glob patterns to exclude files, e.g. `["*.tmp", "**/*.log"]`.
     Any file or directory matching these patterns will be excluded even if they match `included_patterns`.
     If not specified, no files will be excluded.
-*   `sas_token` (`cocoindex.TransientAuthEntryReference[str]`, optional): a SAS token for authentication.
-*   `account_access_key` (`cocoindex.TransientAuthEntryReference[str]`, optional): an account access key for authentication.
 
     :::info
 
     `included_patterns` and `excluded_patterns` are using Unix-style glob syntax. See [globset syntax](https://docs.rs/globset/latest/globset/index.html#syntax) for the details.
 
     :::
 
+*   `max_file_size` (`int`, optional): if provided, files exceeding this size in bytes will be treated as non-existent and skipped during processing.
+    This is useful to avoid processing large files that are not relevant to your use case, such as videos or backups.
+    If not specified, no size limit is applied.
+*   `sas_token` (`cocoindex.TransientAuthEntryReference[str]`, optional): a SAS token for authentication.
+*   `account_access_key` (`cocoindex.TransientAuthEntryReference[str]`, optional): an account access key for authentication.
+
 ### Schema
 
 The output is a [*KTable*](/docs/core/data_types#ktable) with the following sub fields:

python/cocoindex/sources/_engine_builtin_specs.py

@@ -78,6 +78,7 @@ class AzureBlob(op.SourceSpec):
     binary: bool = False
     included_patterns: list[str] | None = None
     excluded_patterns: list[str] | None = None
+    max_file_size: int | None = None
 
     sas_token: TransientAuthEntryReference[str] | None = None
     account_access_key: TransientAuthEntryReference[str] | None = None

src/ops/sources/azure_blob.rs

@@ -19,6 +19,7 @@ pub struct Spec {
     binary: bool,
     included_patterns: Option<Vec<String>>,
     excluded_patterns: Option<Vec<String>>,
+    max_file_size: Option<i64>,
 
     /// SAS token for authentication. Takes precedence over account_access_key.
     sas_token: Option<AuthEntryReference<String>>,
@@ -32,6 +33,7 @@ struct Executor {
     prefix: Option<String>,
     binary: bool,
     pattern_matcher: PatternMatcher,
+    max_file_size: Option<i64>,
 }
 
 fn datetime_to_ordinal(dt: &time::OffsetDateTime) -> Ordinal {
@@ -73,6 +75,15 @@ impl SourceExecutor for Executor {
                     // Only include files (not directories)
                     if key.ends_with('/') { continue; }
 
+                    // Check file size limit
+                    if let Some(max_size) = self.max_file_size {
+                        if let Some(size) = blob.properties.content_length {
+                            if size > max_size {
+                                continue;
+                            }
+                        }
+                    }
+
                     if self.pattern_matcher.is_file_included(key) {
                         let ordinal = Some(datetime_to_ordinal(&blob.properties.last_modified));
                         batch.push(PartialSourceRow {
@@ -115,6 +126,24 @@ impl SourceExecutor for Executor {
             });
         }
 
+        // Check file size limit
+        if let Some(max_size) = self.max_file_size {
+            let blob_client = self
+                .client
+                .container_client(&self.container_name)
+                .blob_client(key_str.as_ref());
+            let properties = blob_client.get_properties().await?;
+            if let Some(size) = properties.blob.properties.content_length {
+                if size > max_size {
+                    return Ok(PartialSourceRowData {
+                        value: Some(SourceValue::NonExistence),
+                        ordinal: Some(Ordinal::unavailable()),
+                        content_version_fp: None,
+                    });
+                }
+            }
+        }
+
         let blob_client = self
             .client
             .container_client(&self.container_name)
@@ -238,6 +267,7 @@ impl SourceFactoryBase for Factory {
             prefix: spec.prefix,
             binary: spec.binary,
             pattern_matcher: PatternMatcher::new(spec.included_patterns, spec.excluded_patterns)?,
+            max_file_size: spec.max_file_size,
         }))
     }
 }