{"record":{"id":"70c2bb4312707ea0","repo":"apache/beam","slug":"snowpipe-is-required-for-streaming-writes","errorCode":null,"errorMessage":"snowPipe is required for streaming writes.","messagePattern":"snowPipe is required for streaming writes\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeWriteSchemaTransformProvider.java","lineNumber":117,"sourceCode":"              configuration.getDatabase(),\n              configuration.getSchema(),\n              configuration.getWarehouse(),\n              configuration.getRole());\n\n      SnowflakeIO.Write<Row> write =\n          SnowflakeIO.<Row>write()\n              .withDataSourceConfiguration(dataSourceConfiguration)\n              .withStagingBucketName(configuration.getStagingBucketName())\n              .withStorageIntegrationName(configuration.getStorageIntegrationName())\n              .withUserDataMapper(row -> row.getValues().toArray());\n\n      boolean streaming = rows.isBounded() == PCollection.IsBounded.UNBOUNDED;\n\n      if (streaming) {\n        String snowPipe = configuration.getSnowPipe();\n\n        if (snowPipe == null || snowPipe.isEmpty()) {\n          throw new IllegalArgumentException(\"snowPipe is required for streaming writes.\");\n        }\n\n        write = write.withSnowPipe(snowPipe);\n\n        Integer flushRowLimit = configuration.getFlushRowLimit();\n        if (flushRowLimit != null) {\n          write = write.withFlushRowLimit(flushRowLimit);\n        }\n\n        Long flushTimeLimitMillis = configuration.getFlushTimeLimitMillis();\n        if (flushTimeLimitMillis != null) {\n          write = write.withFlushTimeLimit(Duration.millis(flushTimeLimitMillis));\n        }\n\n        Integer shardsNumber = configuration.getShardsNumber();\n        if (shardsNumber != null) {\n          write = write.withShardsNumber(shardsNumber);\n        }","sourceCodeStart":99,"sourceCodeEnd":135,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/snowflake/src/main/java/org/apache/beam/sdk/io/snowflake/SnowflakeWriteSchemaTransformProvider.java#L99-L135","documentation":"SnowflakeWriteSchemaTransformProvider.expand() detects an unbounded input PCollection as a streaming write and requires a snowPipe name, because streaming inserts go through a Snowflake SnowPipe. If the input is streaming and snowPipe is null or empty, it throws this IllegalArgumentException at pipeline construction.","triggerScenarios":"Applying the Snowflake write SchemaTransform to an unbounded (streaming, e.g. Kafka-backed) PCollection while the snowPipe configuration field is absent or empty.","commonSituations":"Switching a pipeline from a bounded batch source to a streaming source without adding the SnowPipe config; running the same config in a streaming runner (Flink/Beam streaming) where isBounded() flips to UNBOUNDED.","solutions":["Create a SnowPipe in Snowflake and set it in the configuration, e.g. snowPipe: MY_DB.MY_SCHEMA.MY_PIPE.","If the write is intended to be batch, ensure the input PCollection is bounded (use a bounded source or window/complete the stream).","For batch semantics over a stream, materialize the input (e.g. write to files first) then apply the batch write."],"exampleFix":"// before\nconfig: {serverName: ..., database: ..., } // no snowPipe, streaming input\n// after\nconfig: {serverName: ..., database: ..., snowPipe: \"MY_DB.MY_SCHEMA.MY_PIPE\"}","handlingStrategy":"validation","validationCode":"boolean streaming = pcollection.isBounded() == PCollection.IsBounded.UNBOUNDED;\nif (streaming && (config.getSnowPipe() == null || config.getSnowPipe().isEmpty())) {\n  throw new IllegalArgumentException(\"Streaming input requires snowPipe to be set\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  return applyTransform(input, config);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"snowPipe is required\")) { /* add snowPipe config or make input bounded */ }\n  throw e;\n}","preventionTips":["Always configure snowPipe when a pipeline may run on a streaming runner.","Know your runner's boundedness: Flink/Dataflow streaming make inputs UNBOUNDED.","Create and reference the SnowPipe object in Snowflake before deploying the streaming pipeline."],"tags":["java","snowflake","streaming","configuration"],"backgroundTag":"missing-required-config-field","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}