updated types, made rollover policy more clear, updated path copy, feedback

Author: jonesphillip

packages/wrangler/src/__tests__/pipelines.test.ts

@@ -912,6 +912,7 @@ describe("wrangler pipelines", () => {
 	});
 
 	describe("pipelines streams create", () => {
+		const { setIsTTY } = useMockIsTTY();
 		function mockCreateStreamRequest(expectedRequest: {
 			name: string;
 			hasSchema?: boolean;
@@ -982,6 +983,12 @@ describe("wrangler pipelines", () => {
 		});
 
 		it("should create stream with default settings", async () => {
+			setIsTTY(true);
+			mockConfirm({
+				text: "No schema file provided. Create stream without a schema (unstructured JSON)?",
+				result: true,
+			});
+
 			const createRequest = mockCreateStreamRequest({
 				name: "my_stream",
 			});

packages/wrangler/src/pipelines/cli/setup.ts

@@ -150,6 +150,7 @@ async function setupStreamConfiguration(
 
 	setupConfig.streamConfig = {
 		name: setupConfig.streamName,
+		format: { type: "json" as const, ...(!schema && { unstructured: true }) },
 		http: {
 			enabled: httpEnabled,
 			authentication: httpAuth,
@@ -321,9 +322,19 @@ async function loadSchemaFromFile(): Promise<SchemaField[]> {
 
 		return parsedSchema.fields;
 	} catch (error) {
-		throw new UserError(
+		logger.error(
 			`Failed to read schema file: ${error instanceof Error ? error.message : String(error)}`
 		);
+
+		const retry = await confirm("Would you like to try again?", {
+			defaultValue: true,
+		});
+
+		if (retry) {
+			return loadSchemaFromFile();
+		} else {
+			throw new UserError("Schema file loading cancelled");
+		}
 	}
 }
 
@@ -370,9 +381,12 @@ async function setupR2Sink(
 		);
 	}
 
-	const path = await prompt("File prefix (optional):", {
-		defaultValue: "",
-	});
+	const path = await prompt(
+		"The base prefix in your bucket where data will be written (optional):",
+		{
+			defaultValue: "",
+		}
+	);
 
 	const timePartitionPattern = await prompt(
 		"Time partition pattern (optional):",
@@ -391,7 +405,6 @@ async function setupR2Sink(
 	});
 
 	let compression;
-	let targetRowGroupSize;
 	if (format === "parquet") {
 		compression = await select("Compression:", {
 			choices: [
@@ -401,21 +414,20 @@ async function setupR2Sink(
 				{ title: "zstd", value: "zstd" },
 				{ title: "lz4", value: "lz4" },
 			],
-			defaultOption: 0,
-			fallbackOption: 0,
-		});
-
-		targetRowGroupSize = await prompt("Target row group size (MB):", {
-			defaultValue: "128",
+			defaultOption: 3,
+			fallbackOption: 3,
 		});
 	}
 
-	const fileSizeMB = await prompt("Maximum file size (MB):", {
+	const fileSizeMB = await prompt("Roll file when size reaches (MB):", {
 		defaultValue: "100",
 	});
-	const intervalSeconds = await prompt("Maximum time interval (seconds):", {
-		defaultValue: "300",
-	});
+	const intervalSeconds = await prompt(
+		"Roll file when time reaches (seconds):",
+		{
+			defaultValue: "300",
+		}
+	);
 
 	const useOAuth = await confirm(
 		"Automatically generate credentials needed to write to your R2 bucket?",
@@ -457,9 +469,6 @@ async function setupR2Sink(
 			...(compression && {
 				compression: compression as ParquetFormat["compression"],
 			}),
-			...(targetRowGroupSize && {
-				row_group_bytes: parseInt(targetRowGroupSize) * 1024 * 1024,
-			}),
 		};
 	}
 
@@ -506,24 +515,22 @@ async function setupDataCatalogSink(setupConfig: SetupConfig): Promise<void> {
 		fallbackOption: 0,
 	});
 
-	const targetRowGroupSize = await prompt("Target row group size (MB):", {
-		defaultValue: "128",
-	});
-
-	const fileSizeMB = await prompt("Maximum file size (MB):", {
+	const fileSizeMB = await prompt("Roll file when size reaches (MB):", {
 		defaultValue: "100",
 	});
-	const intervalSeconds = await prompt("Maximum time interval (seconds):", {
-		defaultValue: "300",
-	});
+	const intervalSeconds = await prompt(
+		"Roll file when time reaches (seconds):",
+		{
+			defaultValue: "300",
+		}
+	);
 
 	setupConfig.sinkConfig = {
 		name: setupConfig.sinkName,
 		type: "r2_data_catalog",
 		format: {
 			type: "parquet",
 			compression: compression as ParquetFormat["compression"],
-			row_group_bytes: parseInt(targetRowGroupSize) * 1024 * 1024,
 		},
 		config: {
 			bucket,

packages/wrangler/src/pipelines/cli/sinks/create.ts

@@ -59,23 +59,23 @@ export const pipelinesSinksCreateCommand = createCommand({
 			type: "string",
 		},
 		path: {
-			describe: "Subpath within the bucket",
+			describe: "The base prefix in your bucket where data will be written",
 			type: "string",
 		},
 		partitioning: {
 			describe: "Time partition pattern (r2 sinks only)",
 			type: "string",
 		},
-		"batch-file-size": {
-			describe: "Maximum file size before rolling (e.g., 100MB, 1GB)",
+		"roll-size": {
+			describe: "Roll file when size reaches (e.g., 100MB, 1GB)",
 			type: "string",
 			default:
 				SINK_DEFAULTS.rolling_policy.file_size_bytes === 0
 					? undefined
 					: `${SINK_DEFAULTS.rolling_policy.file_size_bytes}`,
 		},
-		"batch-interval": {
-			describe: "Time interval before rolling (e.g., 5m, 1h)",
+		"roll-interval": {
+			describe: "Roll file when time reaches (e.g., 5m, 1h)",
 			type: "string",
 			default: `${SINK_DEFAULTS.rolling_policy.interval_seconds}s`,
 		},
@@ -175,25 +175,25 @@ export const pipelinesSinksCreateCommand = createCommand({
 			};
 		}
 
-		if (args.batchFileSize || args.batchInterval) {
+		if (args.rollSize || args.rollInterval) {
 			let file_size_bytes: number =
 				SINK_DEFAULTS.rolling_policy.file_size_bytes;
 			let interval_seconds: number =
 				SINK_DEFAULTS.rolling_policy.interval_seconds;
 
-			if (args.batchFileSize) {
+			if (args.rollSize) {
 				// Parse file size (e.g., "100MB" -> bytes)
-				const sizeMatch = args.batchFileSize.match(/^(\d+)(MB|GB)?$/i);
+				const sizeMatch = args.rollSize.match(/^(\d+)(MB|GB)?$/i);
 				if (sizeMatch) {
 					const size = parseInt(sizeMatch[1]);
 					const unit = sizeMatch[2]?.toUpperCase() || "MB";
 					file_size_bytes =
 						unit === "GB" ? size * 1024 * 1024 * 1024 : size * 1024 * 1024;
 				}
 			}
-			if (args.batchInterval) {
+			if (args.rollInterval) {
 				// Parse interval (e.g., "300s" or "5m" -> seconds)
-				const intervalMatch = args.batchInterval.match(/^(\d+)([sm])?$/i);
+				const intervalMatch = args.rollInterval.match(/^(\d+)([sm])?$/i);
 				if (intervalMatch) {
 					const interval = parseInt(intervalMatch[1]);
 					const unit = intervalMatch[2]?.toLowerCase() || "s";

packages/wrangler/src/pipelines/cli/streams/create.ts

@@ -1,5 +1,6 @@
 import { readFileSync } from "node:fs";
 import { createCommand } from "../../../core/create-command";
+import { confirm } from "../../../dialogs";
 import { UserError } from "../../../errors";
 import { logger } from "../../../logger";
 import { parseJSON } from "../../../parse";
@@ -70,11 +71,23 @@ export const pipelinesStreamsCreateCommand = createCommand({
 					`Failed to read or parse schema file '${args.schemaFile}': ${error instanceof Error ? error.message : String(error)}`
 				);
 			}
+		} else {
+			// No schema file provided - confirm with user
+			const confirmNoSchema = await confirm(
+				"No schema file provided. Create stream without a schema (unstructured JSON)?",
+				{ defaultValue: false }
+			);
+
+			if (!confirmNoSchema) {
+				throw new UserError(
+					"Stream creation cancelled. Please provide a schema file using --schema-file"
+				);
+			}
 		}
 
 		const streamConfig: CreateStreamRequest = {
 			name: streamName,
-			...(!schema && { format: { type: "json" as const, unstructured: true } }),
+			format: { type: "json" as const, ...(!schema && { unstructured: true }) },
 			http: {
 				enabled: args.httpEnabled,
 				authentication: args.httpAuth,

packages/wrangler/src/pipelines/cli/streams/utils.ts

@@ -130,10 +130,8 @@ function generateSampleValue(field: SchemaField): SampleValue {
 		case "bool":
 			return true;
 		case "int32":
-		case "u_int32":
 			return 42;
 		case "int64":
-		case "u_int64":
 			return "9223372036854775807"; // Large numbers as strings to avoid JS precision issues
 		case "f32":
 		case "f64":

packages/wrangler/src/pipelines/defaults.ts

@@ -3,8 +3,8 @@ import type { ParquetFormat, Sink } from "./types";
 export const SINK_DEFAULTS = {
 	format: {
 		type: "parquet",
-		compression: "uncompressed",
-		row_group_bytes: 128 * 1024 * 1024,
+		compression: "zstd",
+		row_group_bytes: 1024 * 1024 * 1024,
 	} as ParquetFormat,
 	rolling_policy: {
 		file_size_bytes: 0,

packages/wrangler/src/pipelines/types.ts

@@ -137,8 +137,6 @@ export type SchemaField = {
 		| "bool"
 		| "int32"
 		| "int64"
-		| "u_int32"
-		| "u_int64"
 		| "f32"
 		| "f64"
 		| "string"