diff --git a/tools/bulk_executor/server/src/python_modules/load/__init__.py b/tools/bulk_executor/server/src/python_modules/load/__init__.py index e82d42c7..ac188c96 100644 --- a/tools/bulk_executor/server/src/python_modules/load/__init__.py +++ b/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') diff --git a/tools/bulk_executor/tests/server/test_load.py b/tools/bulk_executor/tests/server/test_load.py index a08ba256..2f4fe5e0 100644 --- a/tools/bulk_executor/tests/server/test_load.py +++ b/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: