{"record":{"id":"fdbde790a66f4e0a","repo":"apache/beam","slug":"unable-run-pipeline-with-create-if-needed-disposition","errorCode":null,"errorMessage":"Unable run pipeline with CREATE IF NEEDED disposition.","messagePattern":"Unable run pipeline with CREATE IF NEEDED disposition\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/services/SnowflakeBatchServiceImpl.java","lineNumber":262,"sourceCode":"          if (!checkResultIfTableExists(resultSet)) {\n            try {\n              createTable(dataSource, table, tableSchema);\n            } catch (SQLException e) {\n              throw new RuntimeException(\"Unable to create table.\", e);\n            }\n          }\n        });\n  }\n\n  private static boolean checkResultIfTableExists(ResultSet resultSet) {\n    try {\n      if (resultSet.next()) {\n        return checkIfResultIsTrue(resultSet);\n      } else {\n        throw new RuntimeException(\"Unable run pipeline with CREATE IF NEEDED - no response.\");\n      }\n    } catch (SQLException e) {\n      throw new RuntimeException(\"Unable run pipeline with CREATE IF NEEDED disposition.\", e);\n    }\n  }\n\n  private void createTable(DataSource dataSource, String table, SnowflakeTableSchema tableSchema)\n      throws SQLException {\n    checkArgument(\n        tableSchema != null,\n        \"The CREATE_IF_NEEDED disposition requires schema if table doesn't exists\");\n    String query = String.format(\"CREATE TABLE %s (%s);\", table, tableSchema.sql());\n    runConnectionWithStatement(dataSource, query, null);\n  }\n\n  private static boolean checkIfResultIsTrue(ResultSet resultSet) throws SQLException {\n    int columnId = 1;\n    return resultSet.getBoolean(columnId);\n  }\n\n  private static void runConnectionWithStatement(","sourceCodeStart":244,"sourceCodeEnd":280,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/services/SnowflakeBatchServiceImpl.java#L244-L280","documentation":"checkResultIfTableExists wraps a SQLException thrown while reading the table-existence result set for the CREATE IF NEEDED disposition. The existence probe failed at the JDBC level, so the pipeline cannot determine whether to create the table.","triggerScenarios":"resultSet.next() or checkIfResultIsTrue throws SQLException during createTableIfNotExists — connection errors, closed result set, invalid cursor state while running the CREATE_IF_NEEDED preflight.","commonSituations":"Transient Snowflake connectivity issues; custom StatementExecutionSupplier closing the result set before Beam reads it; session/warehouse timeouts.","solutions":["Inspect the wrapped SQLException cause for the root JDBC error.","Ensure any custom StatementExecutionSupplier does not close or consume the ResultSet before returning it.","Retry the pipeline; transient Snowflake errors typically clear on retry.","Validate credentials/warehouse state (suspended warehouse, expired session)."],"exampleFix":"// before: supplier consumes the ResultSet\nrs.next(); // read it for logging, then return rs\n// after: return it untouched\nreturn statement.executeQuery(\"SHOW OBJECTS IN \" + tableRef);","handlingStrategy":"retry","validationCode":"// Pre-flight: run the existence probe manually\n// SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='...' AND table_name='...';","typeGuard":null,"tryCatchPattern":"try { pipeline.run(); }\ncatch (RuntimeException e) {\n  if (e.getCause() instanceof java.sql.SQLException sqlEx && isTransient(sqlEx)) retry();\n  else throw e;\n}","preventionTips":["Do not close or iterate the ResultSet before returning it from a StatementExecutionSupplier.","Use stable credentials/warehouse settings to avoid mid-query session drops.","Monitor Snowflake for transient error codes and add retry policy."],"tags":["snowflake","sql","jdbc","result-set"],"backgroundTag":"sql-query-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}