feat: add max_file_size support to AmazonS3 source (#1257)

Add optional max_file_size parameter to filter files by size in both
list() and get_value() APIs. Files exceeding the limit are treated as
non-existent. Closes #1252

Co-authored-by: prabhath004 <ppalakur@gmu.edu>

Author: prabhath004

docs/docs/sources/amazons3.md

@@ -131,6 +131,9 @@ The spec takes the following fields:
 
     :::
 
+*   `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.
 *   `sqs_queue_url` (`str`, optional): if provided, the source will receive change event notifications from Amazon S3 via this SQS queue.
 
     :::info

python/cocoindex/sources/_engine_builtin_specs.py

@@ -55,6 +55,7 @@ class AmazonS3(op.SourceSpec):
     binary: bool = False
     included_patterns: list[str] | None = None
     excluded_patterns: list[str] | None = None
+    max_file_size: int | None = None
     sqs_queue_url: str | None = None
     redis: RedisNotification | None = None
 

src/ops/sources/amazon_s3.rs

@@ -34,6 +34,7 @@ pub struct Spec {
     binary: bool,
     included_patterns: Option<Vec<String>>,
     excluded_patterns: Option<Vec<String>>,
+    max_file_size: Option<i64>,
     sqs_queue_url: Option<String>,
     redis: Option<RedisConfig>,
 }
@@ -82,6 +83,7 @@ struct Executor {
     prefix: Option<String>,
     binary: bool,
     pattern_matcher: PatternMatcher,
+    max_file_size: Option<i64>,
     sqs_context: Option<Arc<SqsContext>>,
     redis_context: Option<Arc<RedisContext>>,
 }
@@ -115,6 +117,14 @@ impl SourceExecutor for Executor {
                         if let Some(key) = obj.key() {
                             // Only include files (not folders)
                             if key.ends_with('/') { continue; }
+                            // Check file size limit
+                            if let Some(max_size) = self.max_file_size {
+                                if let Some(size) = obj.size() {
+                                    if size > max_size {
+                                        continue;
+                                    }
+                                }
+                            }
                             if self.pattern_matcher.is_file_included(key) {
                                 batch.push(PartialSourceRow {
                                     key: KeyValue::from_single_part(key.to_string()),
@@ -156,6 +166,25 @@ impl SourceExecutor for Executor {
                 content_version_fp: None,
             });
         }
+        // Check file size limit
+        if let Some(max_size) = self.max_file_size {
+            let head_result = self
+                .client
+                .head_object()
+                .bucket(&self.bucket_name)
+                .key(key_str.as_ref())
+                .send()
+                .await?;
+            if let Some(size) = head_result.content_length() {
+                if size > max_size {
+                    return Ok(PartialSourceRowData {
+                        value: Some(SourceValue::NonExistence),
+                        ordinal: Some(Ordinal::unavailable()),
+                        content_version_fp: None,
+                    });
+                }
+            }
+        }
         let resp = self
             .client
             .get_object()
@@ -457,6 +486,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,
             sqs_context: spec.sqs_queue_url.map(|url| {
                 Arc::new(SqsContext {
                     client: aws_sdk_sqs::Client::new(&config),