[bulk] Report driver-side failures politely too (closes #332)

tools/bulk_executor/ai_lint/rules/dynamodb_failures_reported.md

@@ -148,15 +148,32 @@ handler exists or is broad enough.
 ## Invariant 2 — driver-side work fails politely too
 
 A verb whose DynamoDB access happens on the **driver** (`find`, `count`, `sql` read
-through `shared/glue_connector`'s `.load()`) has no worker code to check, but it is
-not therefore exempt. A denied `.load()` must still reach the user as one sentence.
-
-The channel is `BulkExecutorError`: `root.py` catches it and calls
-`sys.exit(str(e))`, Glue records that as the job's `ErrorMessage`, and the client
-prints it as its closing line. The PITR guard already rides this path and produces
-exactly one clean sentence. Anything else — a bare Spark/Py4J exception, or a plain
-`Exception` — leaves the user a traceback prefixed with Glue's error category, even
-when the text underneath is perfectly good.
+through `shared/glue_connector`'s `.load()`; `load` writes through it) has no worker code
+to check, but it is not therefore exempt. A denied `.load()` must still reach the user as
+one sentence.
+
+Since #332 there is a net: `root.py` routes everything out of a verb through
+`shared/driver_errors.py`, which classifies it the same way workers are classified and
+**exits rather than re-raising**. Re-raising is what summons Glue's exception-analysis
+blob and Py4J's restatement of the same failure. So a verb no longer *has* to convert its
+own exception for the user to get a sentence — but converting is still better where the
+verb knows something the net cannot, which is why `sql`, `load` and the transform loader
+raise `BulkExecutorError` with their own wording.
+
+Two things to check when reading driver-side code:
+
+- **A marker must reach the console.** An understood failure logs behind
+  `driver_errors.EXPLAINED_FAILURE_PREFIX`, an unexpected one behind
+  `driver_errors.UNEXPECTED_FAILURE_BANNER`, and the client suppresses Glue's blob when it
+  sees either (or `BulkExecutorError`). Glue emits that blob after a clean `sys.exit` only
+  *sometimes* — measured: a denied `find` produced 183 blob lines while a denied `count`
+  in the same batch produced none — so a missing marker is a bug you will only see half
+  the time.
+- **The message has to survive the Java stack.** A Py4J error's `str()` is the whole
+  stack. `get_error_message` unwraps it to the innermost cause, because the outer layer is
+  usually Spark boilerplate ("Job aborted due to stage failure") while the cause is AWS's
+  sentence. If you see a closing line reading `An error occurred while calling o304.load`,
+  that unwrapping did not happen.
 
 Two things that look like compliance and are not:
 
@@ -280,13 +297,11 @@ State what you verified even when clean, so a pass is trustworthy.
 
   Outputs are kept in `~/Documents/bulk-331-runs/` with `before/` counterparts on `main`
   (597-668 lines each).
-- **Does not conform to invariant 2, tracked as #332:** `find`, `count` and `sql`.
-  A table the role cannot `Scan` gives 314-324 lines closing on
-  `Error Category: UNCLASSIFIED_ERROR; Failed Line Number: 1362; An error occurred
-  while calling o304.load. User: ... is not authorized ...`. AWS's sentence is in
-  there, but it arrives as an unhandled Py4J exception. Compare the worker-side
-  verbs, which now close on `Error during delete: User ... is not authorized to
-  perform: dynamodb:BatchWriteItem` in 26-82 lines.
+- Invariant 2 conforms as of #332, via the `root.py` net plus per-site wording in `sql`,
+  `load` and `transform_loader`. Before it, a table the role could not `Scan` gave
+  314-324 lines closing on `Error Category: UNCLASSIFIED_ERROR; ...; An error occurred
+  while calling o304.load. User: ... is not authorized ...`; the sentence was in there,
+  but it arrived as an unhandled Py4J exception.
 - Partly audited: `load`, `load_export`, `revert_export`. Their shared write path
   (`shared/export/pipeline/writer.py`) is now correct, but whether every worker entry
   point in those paths records rather than escapes has not been checked — treat the

tools/bulk_executor/client/src/runner.py

@@ -55,12 +55,16 @@
 ])
 
 # Markers the job prints when it has already told the user what went wrong: a
-# BulkExecutorError sentence, or a worker traceback that names the offending line.
-# Either way Glue's exception analysis adds nothing but volume afterwards.
-# The second must match worker_errors.UNEXPECTED_FAILURE_BANNER.
+# BulkExecutorError sentence, or a traceback that names the offending line -- from a
+# worker or from the driver. Either way Glue's exception analysis adds nothing but volume
+# afterwards, and for a driver-side failure it also restates the same error through Py4J.
+# The last two must match worker_errors.UNEXPECTED_FAILURE_BANNER and
+# driver_errors.UNEXPECTED_FAILURE_BANNER (a guard test checks both).
 JOB_EXPLAINED_THE_FAILURE = (
     'BulkExecutorError',
     'A worker failed in a way we did not expect. Traceback from the worker:',
+    'The job failed in a way we did not expect. Traceback:',
+    'Bulk Executor failure: ',
 )
 
 # Timing constants

tools/bulk_executor/client/src/utils/__init__.py

@@ -38,6 +38,13 @@
     # aws-glue-di-package.jar or metrics-core. Remove once AWS fixes the image.
     # Issue #334.
     r"Exception thrown from AWSDILyraMetricsReporter#report",
+    # Spark reports a query-analysis failure itself, at ERROR, before our handler sees it:
+    # one ~10 KB JSON event carrying the message we go on to print cleanly, plus ~90 Java
+    # frames and the unresolved query plan. Measured on `sql` with a mistyped column: 90 of
+    # the run's 148 lines. Anchored on the logger name inside that JSON, so it drops the
+    # whole event rather than leaving orphaned frames. The message still reaches the user
+    # through the verb's own "SQL query error: ..." line. Issue #332.
+    r'"logger": "SQLQueryContextLogger"',
 ]
 
 # Intentional nuanced configs:

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

@@ -81,20 +81,36 @@ def run(job, spark_context, glue_context, parsed_args):
     if not check_s3_file_exists(s3_path):
         raise BulkExecutorError(f"The S3 path '{s3_path}' doesn't exist or is not accessible")
 
-    dynamicFrame = read_data(glue_context, s3_path, parsed_args)
-
+    # Inside the same handler as the count below: some formats fail here instead. Parquet
+    # reads its footer while the frame is being created, so `--format parquet` at a CSV
+    # file raises from read_data, while a JSON or CSV mismatch only surfaces at count().
+    # Measured before this was wrapped: the Parquet case reported as an unexpected failure
+    # ("Py4JJavaError: ... is not a Parquet file") rather than as the user's own mistake.
     count = 0
     try:
+        dynamicFrame = read_data(glue_context, s3_path, parsed_args)
         count = dynamicFrame.count()
         if count == 0:
-            log.error("No data found, please check your data source") # Should perhaps check that the path exists
+            # 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).
+            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.")
             return
         log.info(f"\nPreparing to load {count} items")
         log.info("Schema is:")
         dynamicFrame.printSchema()
 
     except Exception as e:
-        raise Exception(f"Failed to create DynamicFrame {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
+        # file. (A path that does not exist is caught earlier, before the job starts.)
+        raise BulkExecutorError(
+            f"Could not read the source at '{s3_path}' as {parsed_args.get('format')!r}: "
+            f"{get_error_message(e)}") from None
 
     if parsed_args.get('removeEmptyStringAttributes') is not None:
         log.debug(f"removeEmptyStringAttributes parameter was provided")
@@ -114,7 +130,9 @@ def run(job, spark_context, glue_context, parsed_args):
             glue_context, df, table_name, parsed_args, write_rate=write_rate)
         log.info(f"Wrote {count} items to '{table_name}'")
     except Exception as e:
-        raise Exception(f"Error in writing to table: {get_error_message(e)}") from None
+        # The connector write: a read-only role or persistent throttling lands here, and
+        # 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 check_s3_file_exists(s3_uri):
     """

tools/bulk_executor/server/src/python_modules/shared/driver_errors.py

@@ -0,0 +1,83 @@
+"""Surfacing a failure that happened on the driver.
+
+`worker_errors.py` handles the failures a worker records; this handles the ones the
+driver hits directly -- a denied `.load()` through the Glue connector, a source file
+that isn't the format the user claimed, a typo in `--transform`. The classification is
+the same one, imported from there, because the question is the same: did we understand
+this?
+
+- understood -- a denial, expired credentials, throttling, or something a verb already
+  phrased as a BulkExecutorError. One sentence, no traceback.
+- unexpected -- anything else. The traceback is printed once, on the console, and the
+  job still exits with a one-line reason.
+
+Either way `root.py` exits rather than re-raising, so Glue's exception-analysis blob and
+its Py4J restatement of the same error never reach the user. That matters most for the
+connector: a denied read arrives as a Py4JJavaError whose str() is a 300-line Java stack
+with AWS's sentence buried on line two, and re-raising it means the last thing the user
+sees is `An error occurred while calling o304.load`.
+"""
+
+import sys
+import traceback
+
+from python_modules.shared.bulk_executor_error import BulkExecutorError
+from python_modules.shared.logger import log
+from python_modules.shared.worker_errors import classify_failure
+
+UNEXPECTED_FAILURE_BANNER = "The job failed in a way we did not expect. Traceback:"
+
+# Printed in front of an understood failure's sentence. Two jobs: it tells the reader this
+# is the explanation rather than one more log line, and the client watches for it to
+# suppress Glue's exception-analysis blob. Glue emits that blob for a clean sys.exit only
+# sometimes -- observed on a denied `find` but not a denied `count` in the same batch -- so
+# the marker has to be there every time, not only when a BulkExecutorError was involved.
+#
+# Named rather than generic because the client matches it as a substring anywhere in a log
+# line. "Failure: " alone appeared only in our own output across every run captured for
+# #332, but Spark has shapes like ExecutorLostFailure and FetchFailure that could put
+# "Failure: " in a line of its own -- and a false match would suppress Glue's diagnostics
+# for a failure we had not explained.
+EXPLAINED_FAILURE_PREFIX = "Bulk Executor failure: "
+
+# Cap on what we hand to sys.exit(): Glue records it as the job's ErrorMessage and the
+# client prints it as its closing line. AWS's authorization sentences run ~400 chars,
+# which is worth keeping whole; a Java stack that slipped through is not.
+MAX_REASON_CHARS = 800
+
+
+def _one_line(message):
+    """Collapse to a single line and bound the length. The reason is a closing line, not
+    a report -- anything longer has already been printed above it."""
+    collapsed = " ".join(str(message).split())
+    if len(collapsed) > MAX_REASON_CHARS:
+        collapsed = collapsed[:MAX_REASON_CHARS - 3] + "..."
+    return collapsed
+
+
+def surface(exception):
+    """Report a driver-side failure and exit. Never returns.
+
+    Called by root.py for anything that escapes a verb.
+    """
+    understood, message = classify_failure(exception)
+    if not understood:
+        # Ours to debug, or a user's generator/transform: the frames are the report, and
+        # printing them ourselves means the client suppresses Glue's blob (it watches for
+        # this banner and for BulkExecutorError) and the closing line stays short.
+        print(UNEXPECTED_FAILURE_BANNER)
+        print(traceback.format_exc())
+        # classify_failure already extracted the readable message; the type goes in front
+        # because an unexpected message alone can be as bare as "'pk'".
+        reason = f"{type(exception).__name__}: {message}"
+    else:
+        reason = message
+
+    reason = _one_line(reason)
+    if isinstance(exception, BulkExecutorError):
+        # Keep the name users have seen since before any of this, and the client's
+        # long-standing suppression marker.
+        log.error(f"BulkExecutorError: {reason}")
+    else:
+        log.error(f"{EXPLAINED_FAILURE_PREFIX}{reason}")
+    sys.exit(reason)

tools/bulk_executor/server/src/python_modules/shared/errors.py

@@ -47,14 +47,36 @@ def get_error_message(e):
         except json.JSONDecodeError:
             pass  # fallback
 
-    # Look for DynamoDB exception message
+    # Look for DynamoDB exception message (AWS SDK v1, which Glue 4 and earlier used)
     dynamo_match = re.search(
         r'com\.amazonaws\.services\.dynamodbv2\.model\.AmazonDynamoDBException:\s*(.*?)\s*\(Service:',
         msg
     )
     if dynamo_match:
         return dynamo_match.group(1).strip()
 
+    # A Java exception relayed through Py4J. str() on one of these is the entire Java
+    # stack -- hundreds of frames -- with the sentence that matters on the second line:
+    #
+    #   py4j.protocol.Py4JJavaError: An error occurred while calling o304.load.
+    #   : software.amazon.awssdk.services.dynamodb.model.DynamoDbException: User: ...
+    #           is not authorized to perform: dynamodb:Scan on resource: ...
+    #           at software.amazon.awssdk...(DynamoDbException.java:113)
+    #
+    # Take the innermost cause's message: for a wrapped failure the outer layer is
+    # usually Spark boilerplate ("Job aborted due to stage failure") and the cause is
+    # the AWS sentence the user needs. The message can wrap onto continuation lines, so
+    # keep going until a stack frame or a new exception header.
+    java_causes = re.findall(
+        r'^(?:: |Caused by: )(?:[\w$]+\.)+([\w$]*(?:Exception|Error)): '
+        r'(.*(?:\n(?!\s*(?:at |\.\.\. )|: |Caused by: ).*)*)',
+        msg, re.MULTILINE)
+    if java_causes:
+        _cls, detail = java_causes[-1]
+        detail = ' '.join(detail.split())
+        if detail:
+            return detail
+
     # ParseException handling
     if hasattr(e, 'desc'):  # ParseException
         msg = e.desc

tools/bulk_executor/server/src/python_modules/shared/export/pipeline/transform_loader.py

@@ -1,4 +1,5 @@
 import importlib
+from ...bulk_executor_error import BulkExecutorError
 
 
 def load_transform_module(module_name, transform_package):
@@ -18,4 +19,7 @@ def load_transform_module(module_name, transform_package):
     try:
         return importlib.import_module(f"{transform_package}.{module_name}")
     except ImportError as e:
-        raise ImportError(f"Cannot import transform module '{module_name}' from '{transform_package}': {e}")
+        # --transform is user input, so a typo is a sentence rather than a traceback.
+        raise BulkExecutorError(
+            f"Cannot import transform module '{module_name}' from '{transform_package}': {e}"
+        ) from None

tools/bulk_executor/server/src/python_modules/sql.py

@@ -92,7 +92,8 @@ def run(job, spark_context, glue_context, parsed_args):
         # without wrapping them into an opaque "SQL query error".
         raise
     except Exception as e:
-        raise Exception("SQL query error: " + get_error_message(e)) from None
+        # The query is the user's to fix, so this is a sentence, not a stack trace.
+        raise BulkExecutorError("SQL query error: " + get_error_message(e)) from None
     finally:
         # Ensure Spark session cleanup
         try:

tools/bulk_executor/server/src/root.py

@@ -7,6 +7,7 @@
 from awsglue.job import Job
 from awsglue.transforms import *
 from pyspark.context import SparkContext
+from python_modules.shared import driver_errors
 from python_modules.shared.bulk_executor_error import BulkExecutorError
 
 
@@ -103,11 +104,21 @@ def _get_parsed_glue_job_args(argv):
         action_script_function = getattr(module, action_script_function_name)
         try:
             action_script_function(job, spark_context, glue_context, parsed_args)  # Run the function
-        except BulkExecutorError as e:
-            log.error(f"BulkExecutorError: {e}")
-            sys.exit(str(e))
-        except Exception as e:
-            raise # Just let it propagate
+        except BaseException as e:
+            # Everything a verb can fail with lands here, and shared/driver_errors.py
+            # decides what the user sees: a denial or other understood failure exits with
+            # one sentence, anything else prints its traceback first and then exits with a
+            # one-line reason. Exiting rather than re-raising is the point -- a re-raise
+            # hands the user Glue's exception-analysis blob plus Py4J's restatement of the
+            # same error, with AWS's actual sentence buried inside a Java stack (#332).
+            #
+            # BaseException rather than Exception so nothing escapes by inheriting from
+            # the wrong base -- but SystemExit passes straight through: a helper that
+            # already called exit() has said its piece, and re-reporting it would relabel
+            # a deliberate exit as a surprise.
+            if isinstance(e, SystemExit):
+                raise
+            driver_errors.surface(e)
     else:
         raise Exception(f"Could not find the function '{action_script_function_name}' within the module '{module_name}'.")