Skip to content

Updated catalog compaction arg, small pipelines fixes #10743

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
petebacondarwin merged 2 commits into from
Sep 24, 2025
Merged

Updated catalog compaction arg, small pipelines fixes #10743

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bca4c9f
Updated catalog compaction enable argument from targetSizeMb to targe…
jonesphillip Sep 24, 2025
7f68455
Small fixes for pipelines commands, defaults.
jonesphillip Sep 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/brave-memes-raise.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"wrangler": patch
---

Changes fileSizeMB->file-size for compaction arg. Small fixes for pipelines commands
6 changes: 3 additions & 3 deletions packages/wrangler/src/__tests__/pipelines.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1366,7 +1366,7 @@ describe("wrangler pipelines", () => {

Batching:
File Size: none
Time Interval: 30s
Time Interval: 300s

Format:
Type: json"
Expand Down Expand Up @@ -1400,7 +1400,7 @@ describe("wrangler pipelines", () => {

Batching:
File Size: none
Time Interval: 30s
Time Interval: 300s

Format:
Type: json"
Expand Down Expand Up @@ -1564,7 +1564,7 @@ describe("wrangler pipelines", () => {

Batching:
File Size: none
Time Interval: 30s
Time Interval: 300s

Format:
Type: json"
Expand Down
8 changes: 4 additions & 4 deletions packages/wrangler/src/__tests__/r2.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1301,7 +1301,7 @@ For more details, refer to: https://developers.cloudflare.com/r2/api/s3/tokens/"
)
);
await runWrangler(
"r2 bucket catalog compaction enable testBucket --token fakecloudflaretoken --targetSizeMb 512"
"r2 bucket catalog compaction enable testBucket --token fakecloudflaretoken --target-size 512"
);
expect(std.out).toMatchInlineSnapshot(
`"✨ Successfully enabled file compaction for the data catalog for bucket 'testBucket'."`
Expand Down Expand Up @@ -1332,8 +1332,8 @@ For more details, refer to: https://developers.cloudflare.com/r2/api/s3/tokens/"
-v, --version Show version number [boolean]

OPTIONS
--targetSizeMb The target size for compacted files (allowed values: 64, 128, 256, 512) [number] [default: 128]
--token A cloudflare api token with access to R2 and R2 Data Catalog which will be used to read/write files for compaction. [string] [required]"
--target-size The target size for compacted files in MB (allowed values: 64, 128, 256, 512) [number] [default: 128]
--token A cloudflare api token with access to R2 and R2 Data Catalog which will be used to read/write files for compaction. [string] [required]"
`);
expect(std.err).toMatchInlineSnapshot(`
"X [ERROR] Not enough non-option arguments: got 0, need at least 1
Expand All @@ -1345,7 +1345,7 @@ For more details, refer to: https://developers.cloudflare.com/r2/api/s3/tokens/"
it("should error if --token is not provided", async () => {
await expect(
runWrangler(
"r2 bucket catalog compaction enable testBucket --targetSizeMb 512"
"r2 bucket catalog compaction enable testBucket --target-size 512"
)
).rejects.toThrowErrorMatchingInlineSnapshot(
`[Error: Missing required argument: token]`
Expand Down
37 changes: 21 additions & 16 deletions packages/wrangler/src/pipelines/cli/setup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import {
deleteStream,
validateSql,
} from "../client";
import { SINK_DEFAULTS } from "../defaults";
import { authorizeR2Bucket } from "../index";
import {
displayUsageExamples,
Expand Down Expand Up @@ -231,10 +232,8 @@ async function buildField(
{ title: "string", value: "string" },
{ title: "int32", value: "int32" },
{ title: "int64", value: "int64" },
{ title: "u_int32", value: "u_int32" },
{ title: "u_int64", value: "u_int64" },
{ title: "f32", value: "f32" },
{ title: "f64", value: "f64" },
{ title: "float32", value: "float32" },
{ title: "float64", value: "float64" },
{ title: "bool", value: "bool" },
{ title: "timestamp", value: "timestamp" },
{ title: "json", value: "json" },
Expand Down Expand Up @@ -419,13 +418,16 @@ async function setupR2Sink(
});
}

const fileSizeMB = await prompt("Roll file when size reaches (MB):", {
defaultValue: "100",
});
const fileSizeMB = await prompt(
"Roll file when size reaches (MB, minimum 5):",
{
defaultValue: "100",
}
);
const intervalSeconds = await prompt(
"Roll file when time reaches (seconds):",
"Roll file when time reaches (seconds, minimum 10):",
{
defaultValue: "300",
defaultValue: String(SINK_DEFAULTS.rolling_policy.interval_seconds),
}
);

Expand Down Expand Up @@ -511,17 +513,20 @@ async function setupDataCatalogSink(setupConfig: SetupConfig): Promise<void> {
{ title: "zstd", value: "zstd" },
{ title: "lz4", value: "lz4" },
],
defaultOption: 0,
fallbackOption: 0,
defaultOption: 3,
fallbackOption: 3,
});

const fileSizeMB = await prompt("Roll file when size reaches (MB):", {
defaultValue: "100",
});
const fileSizeMB = await prompt(
"Roll file when size reaches (MB, minimum 5):",
{
defaultValue: "100",
}
);
const intervalSeconds = await prompt(
"Roll file when time reaches (seconds):",
"Roll file when time reaches (seconds, minimum 10):",
{
defaultValue: "300",
defaultValue: String(SINK_DEFAULTS.rolling_policy.interval_seconds),
}
);

Expand Down
5 changes: 2 additions & 3 deletions packages/wrangler/src/pipelines/cli/sinks/create.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,6 @@ export const pipelinesSinksCreateCommand = createCommand({
"roll-size": {
describe: "Roll file size in MB",
type: "number",
default: SINK_DEFAULTS.rolling_policy.file_size_bytes / (1024 * 1024),
},
"roll-interval": {
describe: "Roll file interval in seconds",
Expand Down Expand Up @@ -183,7 +182,7 @@ export const pipelinesSinksCreateCommand = createCommand({
}

if (args.rollSize || args.rollInterval) {
let file_size_bytes: number =
let file_size_bytes: number | undefined =
SINK_DEFAULTS.rolling_policy.file_size_bytes;
let interval_seconds: number =
SINK_DEFAULTS.rolling_policy.interval_seconds;
Expand All @@ -196,7 +195,7 @@ export const pipelinesSinksCreateCommand = createCommand({
}

sinkConfig.config.rolling_policy = {
file_size_bytes,
...(file_size_bytes !== undefined && { file_size_bytes }),
interval_seconds,
};
}
Expand Down
2 changes: 1 addition & 1 deletion packages/wrangler/src/pipelines/cli/sinks/utils.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ export function displaySinkConfiguration(

const batching: Record<string, string> = {
"File Size":
fileSizeBytes === 0
fileSizeBytes === undefined || fileSizeBytes === 0
? "none"
: `${Math.round(fileSizeBytes / (1024 * 1024))}MB`,
"Time Interval": `${intervalSeconds}s`,
Expand Down
4 changes: 2 additions & 2 deletions packages/wrangler/src/pipelines/cli/streams/utils.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,8 +135,8 @@ function generateSampleValue(field: SchemaField): SampleValue {
return 42;
case "int64":
return "9223372036854775807"; // Large numbers as strings to avoid JS precision issues
case "f32":
case "f64":
case "float32":
case "float64":
return 3.14;
case "json":
return { example: "json_value" };
Expand Down
15 changes: 11 additions & 4 deletions packages/wrangler/src/pipelines/defaults.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@ export const SINK_DEFAULTS = {
row_group_bytes: 1024 * 1024 * 1024,
} as ParquetFormat,
rolling_policy: {
file_size_bytes: 0,
interval_seconds: 30,
file_size_bytes: undefined,
interval_seconds: 300,
},
r2: {
path: "",
Expand Down Expand Up @@ -38,11 +38,18 @@ export function applyDefaultsToSink(sink: Sink): Sink {

if (!withDefaults.config.rolling_policy) {
withDefaults.config.rolling_policy = {
file_size_bytes: SINK_DEFAULTS.rolling_policy.file_size_bytes,
interval_seconds: SINK_DEFAULTS.rolling_policy.interval_seconds,
};
// Only add file_size_bytes if it has a value
if (SINK_DEFAULTS.rolling_policy.file_size_bytes !== undefined) {
withDefaults.config.rolling_policy.file_size_bytes =
SINK_DEFAULTS.rolling_policy.file_size_bytes;
}
} else {
if (!withDefaults.config.rolling_policy.file_size_bytes) {
if (
!withDefaults.config.rolling_policy.file_size_bytes &&
SINK_DEFAULTS.rolling_policy.file_size_bytes !== undefined
) {
withDefaults.config.rolling_policy.file_size_bytes =
SINK_DEFAULTS.rolling_policy.file_size_bytes;
}
Expand Down
8 changes: 4 additions & 4 deletions packages/wrangler/src/pipelines/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,8 +137,8 @@ export type SchemaField = {
| "bool"
| "int32"
| "int64"
| "f32"
| "f64"
| "float32"
| "float64"
| "string"
| "timestamp"
| "json"
Expand Down Expand Up @@ -169,7 +169,7 @@ export type Sink = {
time_pattern: string;
};
rolling_policy?: {
file_size_bytes: number;
file_size_bytes?: number;
interval_seconds: number;
Comment on lines +172 to 173
Copy link
Contributor

@petebacondarwin petebacondarwin Sep 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just noting that file size is now optional but interval is not?

Copy link
Contributor Author

@jonesphillip jonesphillip Sep 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes that's correct, since interval_seconds has a default so should always be present

};
// r2_data_catalog specific fields
Expand Down Expand Up @@ -207,7 +207,7 @@ export interface CreateSinkRequest {
time_pattern: string;
};
rolling_policy?: {
file_size_bytes: number;
file_size_bytes?: number;
interval_seconds: number;
};
// R2 credentials (for r2 type)
Expand Down
6 changes: 3 additions & 3 deletions packages/wrangler/src/r2/catalog.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,9 +181,9 @@ export const r2BucketCatalogCompactionEnableCommand = createCommand({
type: "string",
demandOption: true,
},
targetSizeMb: {
"target-size": {
describe:
"The target size for compacted files (allowed values: 64, 128, 256, 512)",
"The target size for compacted files in MB (allowed values: 64, 128, 256, 512)",
type: "number",
demandOption: false,
default: 128,
Copy link
Contributor

@petebacondarwin petebacondarwin Sep 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pretty sure that by having a default here, this option will never be undefined which might mess up your logic later?

Copy link
Contributor

@petebacondarwin petebacondarwin Sep 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I might be totally wrong but in other parts of Wrangler we have a pattern where a value can be configured in both the Wrangler config file but also on the command line.
In that case it is important that we don't have a default for the command line arg because then we never know if the user actually specified a value or whether it is just the default, so we don't know whether to override what is in the config.

Copy link
Contributor

@petebacondarwin petebacondarwin Sep 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK I am totally wrong. I see that this option doesn't get merged with other config. So it is all good.

Copy link
Contributor

@petebacondarwin petebacondarwin Sep 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I note that now this value is used in the only call to enabledR2CatalogCompaction(), which expects the value of targetSizeMb to be number|undefined.

But since this will always be defined (thanks to the default value) then that function could just take number.
The follow on is that R2CatalogCompactionConfig can make its targetSizMb to be only number. Making this comment redundant:

// If undefined, the service will set the default value

Expand All @@ -209,7 +209,7 @@ export const r2BucketCatalogCompactionEnableCommand = createCommand({
config,
accountId,
args.bucket,
args.targetSizeMb
args.targetSize
);

logger.log(
Expand Down
Loading