[bulk] Tell an empty source apart from one load could not read (closes #340)

tools/bulk_executor/server/src/python_modules/load/__init__.py

@@ -91,19 +91,33 @@ def run(job, spark_context, glue_context, parsed_args):
         dynamicFrame = read_data(glue_context, s3_path, parsed_args)
         count = dynamicFrame.count()
         if count == 0:
-            # Logged as a warning, not an error: this run goes on to succeed, and an ERROR
-            # line above "Job completed successfully" is a contradiction the reader has to
-            # resolve. An empty drop is also a legitimate input -- the export pipeline
-            # treats a 0-item export the same way. Whether a source that holds bytes but
-            # yields no rows should fail instead is a separate question (#340).
+            # Zero rows means one of two different things, and they deserve different
+            # outcomes (#340). An empty drop is a legitimate input -- the export pipeline
+            # treats a 0-item export the same way -- so it succeeds with a warning rather
+            # than an ERROR line above "Job completed successfully". A source that holds
+            # bytes and yields nothing is the reader failing to understand it, which is the
+            # user's mistake and worth failing on: otherwise a mistyped --format is
+            # indistinguishable from an empty file, and Spark's JSON reader returns no rows
+            # where its Parquet reader would have raised.
+            source_bytes = s3_source_bytes(s3_path)
+            if source_bytes:
+                raise BulkExecutorError(
+                    f"Read 0 items from '{s3_path}', but it holds {source_bytes:,} bytes. "
+                    f"The data is probably not {parsed_args.get('format')!r} -- check "
+                    f"--format. (A header-only CSV also reads as 0 items.)")
             log.warning(
-                f"Read 0 items from '{s3_path}' as {parsed_args.get('format')!r} -- "
-                f"nothing was loaded. If that is unexpected, check --format and the path.")
+                f"Read 0 items from '{s3_path}': the source is empty, so nothing was "
+                f"loaded. Check the path if that is unexpected.")
             return
         log.info(f"\nPreparing to load {count} items")
         log.info("Schema is:")
         dynamicFrame.printSchema()
 
+    except BulkExecutorError:
+        # Already phrased -- the zero-row branch above raises from inside this try, and
+        # wrapping it again produced "Could not read the source ... as 'json': Read 0 items
+        # from ..., but it holds 29 bytes ..." in a live run.
+        raise
     except Exception as e:
         # This is where Spark actually reads the source, so it fires on the everyday
         # mistakes: --format json pointed at CSV, malformed JSON, an unreadable Parquet
@@ -134,6 +148,35 @@ def run(job, spark_context, glue_context, parsed_args):
         # get_error_message unwraps the Py4J stack to AWS's own sentence.
         raise BulkExecutorError(f"Error in writing to table: {get_error_message(e)}") from None
 
+def _split_s3_uri(s3_uri):
+    """Return (bucket, key) for an s3:// URI."""
+    match = re.match(r"s3://([^/]+)/(.*)", s3_uri)
+    if not match:
+        raise BulkExecutorError(f"Invalid S3 URI format: {s3_uri}. Expected format: s3://bucket-name/key")
+    return match.group(1), match.group(2)
+
+
+def s3_source_bytes(s3_uri):
+    """Total bytes behind the path: one object, or every object under a prefix.
+
+    Only called when a read produced no rows, so the happy path pays nothing for it. The
+    number separates "there was nothing to load" from "the reader did not understand what
+    is there", which otherwise look identical (#340).
+    """
+    bucket_name, key = _split_s3_uri(s3_uri)
+    s3 = boto3.client('s3')
+    try:
+        return s3.head_object(Bucket=bucket_name, Key=key)['ContentLength']
+    except ClientError as e:
+        if e.response['Error']['Code'] != '404':
+            raise
+
+    total = 0
+    for page in s3.get_paginator('list_objects_v2').paginate(Bucket=bucket_name, Prefix=key):
+        total += sum(obj['Size'] for obj in page.get('Contents', []))
+    return total
+
+
 def check_s3_file_exists(s3_uri):
     """
     Check if a specific file exists in S3 using an S3 URI
@@ -144,15 +187,7 @@ def check_s3_file_exists(s3_uri):
     Returns:
         bool: True if the file exists, False otherwise
     """
-    # Parse the S3 URI to extract bucket name and key
-    uri_pattern = r"s3://([^/]+)/(.*)"
-    match = re.match(uri_pattern, s3_uri)
-
-    if not match:
-        raise BulkExecutorError(f"Invalid S3 URI format: {s3_uri}. Expected format: s3://bucket-name/key")
-
-    bucket_name = match.group(1)
-    key = match.group(2)
+    bucket_name, key = _split_s3_uri(s3_uri)
 
     # Initialize S3 client
     s3 = boto3.client('s3')

tools/bulk_executor/tests/server/test_load.py

@@ -262,6 +262,7 @@ class TestRunDynamicFrameCount:
     def test_returns_early_when_count_is_zero(self, monkeypatch):
         """Line 92-94: zero items -> early return with error log."""
         monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True)
+        monkeypatch.setattr(load_module, 's3_source_bytes', lambda uri: 0)
         monkeypatch.setattr(load_module, 'get_dynamodb_throughput_configs',
                             lambda *a, **kw: {})
         glue_ctx = MagicMock()
@@ -280,6 +281,8 @@ def test_zero_rows_warns_and_succeeds(self, monkeypatch, caplog):
         import logging
 
         monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True)
+        # an empty source: nothing to load, which is not the user's mistake
+        monkeypatch.setattr(load_module, 's3_source_bytes', lambda uri: 0)
         df = MagicMock()
         df.count.return_value = 0
         monkeypatch.setattr(load_module, 'read_data', lambda *a: df)
@@ -293,10 +296,37 @@ def test_zero_rows_warns_and_succeeds(self, monkeypatch, caplog):
         levels = {r.levelname for r in caplog.records}
         assert 'ERROR' not in levels, "a successful run must not log an error"
         message = ' '.join(r.message for r in caplog.records)
-        assert 'Read 0 items' in message and "'json'" in message and '--format' in message, \
+        assert 'Read 0 items' in message and 'source is empty' in message, \
             "say what happened and what to check"
         write.assert_not_called()
 
+    def test_zero_rows_from_a_source_with_bytes_is_the_users_mistake(self, monkeypatch):
+        """The wrong-format case. Spark's JSON reader returns no rows where its Parquet
+        reader raises, so without this the same operator error is a clean failure in one
+        format and a silent success in another (#340)."""
+        monkeypatch.setattr(load_module, 'check_s3_file_exists', lambda uri: True)
+        monkeypatch.setattr(load_module, 's3_source_bytes', lambda uri: 4096)
+        df = MagicMock()
+        df.count.return_value = 0
+        monkeypatch.setattr(load_module, 'read_data', lambda *a: df)
+        write = MagicMock()
+        monkeypatch.setattr(load_module, 'write_dynamodb_dataframe', write)
+
+        with pytest.raises(load_module.BulkExecutorError) as exc:
+            load_module.run(MagicMock(), MagicMock(), MagicMock(),
+                            {'table': 't', 's3_path': 's3://b/k/data.json', 'format': 'json'})
+
+        message = str(exc.value)
+        assert not message.startswith('Could not read the source'), (
+            "the zero-row message must not be re-wrapped by the read handler it raises inside; "
+            "a live run produced both messages nested"
+        )
+        assert message.count('Read 0 items') == 1
+        assert '4,096 bytes' in message, "the byte count is the evidence"
+        assert "'json'" in message and '--format' in message
+        assert 'header-only CSV' in message, "name the false positive rather than hide it"
+        write.assert_not_called()
+
     def test_raises_when_the_frame_cannot_even_be_created(self, monkeypatch):
         """Parquet reads its footer while the frame is created, so `--format parquet` at a
         CSV file raises from read_data rather than from count(). Both are the same mistake
@@ -549,6 +579,59 @@ def test_no_rate_when_throughput_configs_returns_empty(self, monkeypatch):
         assert mock_write.call_args.kwargs['write_rate'] is None
 
 
+# --- s3_source_bytes --------------------------------------------------------
+
+class TestS3SourceBytes:
+    """How much data is behind the path (#340). Called only when a read produced no rows,
+    which is why it may cost an API call at all."""
+
+    def test_single_object_uses_head_object(self, monkeypatch):
+        s3 = MagicMock()
+        s3.head_object.return_value = {'ContentLength': 4096}
+        monkeypatch.setattr(load_module.boto3, 'client', lambda *a, **kw: s3)
+
+        assert load_module.s3_source_bytes('s3://bucket/data.json') == 4096
+        s3.get_paginator.assert_not_called(), "one object needs no listing"
+
+    def test_prefix_sums_every_object_across_pages(self, monkeypatch):
+        from botocore.exceptions import ClientError
+        s3 = MagicMock()
+        s3.head_object.side_effect = ClientError(
+            {'Error': {'Code': '404', 'Message': 'Not Found'}}, 'HeadObject')
+        s3.get_paginator.return_value.paginate.return_value = [
+            {'Contents': [{'Size': 10}, {'Size': 20}]},
+            {'Contents': [{'Size': 5}]},
+        ]
+        monkeypatch.setattr(load_module.boto3, 'client', lambda *a, **kw: s3)
+
+        assert load_module.s3_source_bytes('s3://bucket/prefix/') == 35
+
+    def test_empty_prefix_is_zero_bytes(self, monkeypatch):
+        from botocore.exceptions import ClientError
+        s3 = MagicMock()
+        s3.head_object.side_effect = ClientError(
+            {'Error': {'Code': '404', 'Message': 'Not Found'}}, 'HeadObject')
+        s3.get_paginator.return_value.paginate.return_value = [{}]
+        monkeypatch.setattr(load_module.boto3, 'client', lambda *a, **kw: s3)
+
+        assert load_module.s3_source_bytes('s3://bucket/prefix/') == 0
+
+    def test_other_s3_errors_are_not_swallowed(self, monkeypatch):
+        """A denial here must not be reported as "the source is empty"."""
+        from botocore.exceptions import ClientError
+        s3 = MagicMock()
+        s3.head_object.side_effect = ClientError(
+            {'Error': {'Code': 'AccessDenied', 'Message': 'nope'}}, 'HeadObject')
+        monkeypatch.setattr(load_module.boto3, 'client', lambda *a, **kw: s3)
+
+        with pytest.raises(ClientError):
+            load_module.s3_source_bytes('s3://bucket/data.json')
+
+    def test_malformed_uri(self):
+        with pytest.raises(load_module.BulkExecutorError, match="Invalid S3 URI format"):
+            load_module.s3_source_bytes('not-an-s3-uri')
+
+
 # --- check_s3_file_exists ---------------------------------------------------
 
 class TestCheckS3FileExists: