apache/beam · error · RuntimeException

Error mapping Neo4J result

Error message

Error mapping Neo4J result

What it means

Inside ReadFn's transaction work, each Result record is converted via the user-supplied rowMapper. If mapRow throws for any record (bad field access, wrong type, ClassCastException in extraction), the DoFn wraps it in a RuntimeException to add context and abort the transaction, causing the whole read to retry/fail.

Solutions

1. Align the rowMapper with the exact RETURN clause: use record.get("alias") with aliases matching the Cypher.
2. Defensively check record.get("field").isNull() / containsKey before typed extraction.
3. Catch and log per-record mapping issues inside the mapper if partial results are acceptable; otherwise fix the query/mismatch and rerun.

Example fix

// before
record -> record.get("person_name").asString()
// cypher returns: RETURN n.name AS name
// after
record -> {
  Value v = record.get("name");
  return v.isNull() ? null : v.asString();
}

// Validate the mapper against the query result in a test: try (Result r = session.run(cypher)) { rowMapper.mapRow(r.next()); }

try { output(rowMapper.mapRow(record)); } catch (Exception e) { LOG.error("mapRow failed for record %s", record); throw new RuntimeException("Error mapping Neo4J result", e); }

When it happens

Trigger: rowMapper accesses a column not present in the Cypher result (typo or changed RETURN clause), or calls typed getters (asInt/asString) on values of a different or NULL type.

Common situations: Cypher query edited to rename/alias a returned column while the mapper still uses the old name; nodes with missing properties returning NullValue; type changes after schema/DB migration.

Related errors

• Error writing " + unwindList.size() + " rows to Neo4j with…
• neo4j session was not initialized correctly
• null driver given by driver provider
• please provide a parameters function
• please provide a row mapper

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/3adf9cc048775d35. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/io/neo4j/src/main/java/org/apache/beam/sdk/io/neo4j/Neo4jIO.java:825

      }
      executeReadCypherStatement(context, parametersMap);
    }

    private void executeReadCypherStatement(
        final ProcessContext processContext, Map<String, Object> parametersMap) {
      // The actual "reading" work needs to happen in a transaction.
      // We could actually read and write here depending on the type of transaction
      // we picked.  As long as the Cypher statement returns values it's fine.
      //
      TransactionWork<Long> transactionWork =
          transaction -> {
            long count = 0L;
            Result result = transaction.run(cypher, parametersMap);
            while (result.hasNext()) {
              try {
                processContext.output(rowMapper.mapRow(result.next()));
              } catch (Exception e) {
                throw new RuntimeException("Error mapping Neo4J result", e);
              }
              count++;
            }
            return count;
          };

      if (logCypher) {
        String parametersString = getParametersString(parametersMap);

        String readWrite = writeTransaction ? "write" : "read";
        LOG.info(
            "Starting a {} transaction for cypher: {}, parameters: {}",
            readWrite,
            cypher,
            parametersString);
      }

      // There are 2 ways to do a transaction on Neo4j: read or write

View on GitHub (pinned to 12126d8942)