pierrejeambrun commented on code in PR #73723:
URL: https://github.com/apache/airflow/pull/73723#discussion_r4105491744
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -42,27 +44,45 @@ import type { BundleManifest } from
"../coordinator/manifest.js";
const AIRFLOW_BUNDLE_METADATA_VERSION = "1.0";
const EMBEDDED_METADATA_MAX_BYTES = 1024 * 1024;
-const EMBEDDED_SOURCE_MAX_BYTES = 1024 * 1024;
+const EMBEDDED_SOURCES_MAX_BYTES = 4 * 1024 * 1024;
const OFFSET_HEX_WIDTH = 16;
export const EMBEDDED_METADATA_PREFIX = "//# airflowMetadata=";
export const EMBEDDED_LAYOUT_PREFIX = "//# airflowBundle=";
-/** A block comment, because the entrypoint spans more than one line. */
-export const EMBEDDED_SOURCE_OPEN = "/*# airflowSource\n";
+/**
+ * Each source is wrapped in a block comment. The path follows the marker so
+ * a reader can tell one region from another without cross-referencing the
+ * header — the header's byte ranges are the source of truth, but a human
+ * scanning the bundle can still find their file by name.
+ */
+export const EMBEDDED_SOURCE_OPEN_PREFIX = "/*# airflowSource:";
+export const EMBEDDED_SOURCE_OPEN_SUFFIX = "\n";
export const EMBEDDED_SOURCE_CLOSE = "\n#*/\n";
export interface BundleEncoderInput {
bundleManifest: BundleManifest;
sdkVersion: string;
+ /** The entrypoint the packer was invoked on; recorded in metadata so a
reader can
+ * point at the primary source file when there is no per-Dag source path to
use. */
entrypointName: string;
- entrypointSource: string;
+ /**
+ * Every author-owned source file that declares at least one Dag, keyed by
+ * the path used in `BundleManifest.dag_source_paths`. Files that only
+ * import from these (utilities, types) are not embedded — the Code tab
+ * reads what defines each Dag, not what it depends on.
+ *
+ * A bundle whose Dags are all task handlers (Python owns them) may have
+ * an empty map here.
+ */
+ entrypointSources: Record<string, string>;
executable: Uint8Array;
}
interface BundleMetadata {
airflow_bundle_metadata_version: string;
sdk: { language: string; version: string; supervisor_schema_version: string
};
- source: string;
+ entrypoint: string;
+ dag_source_paths: BundleManifest["dag_source_paths"];
Review Comment:
Deliberate — the SDK is at 0.1.0-beta1 and the bundle format has no shipped
consumers yet. The reader-side changes for these fields land alongside the
importer PR. Bumping the metadata version before there's a reader that has to
distinguish old from new would be back-compat theater.
##########
ts-sdk/src/cli/bundle-encoder.ts:
##########
@@ -72,65 +92,86 @@ interface VerifiedByteRange {
start: string;
}
+interface VerifiedSourceRegion extends VerifiedByteRange {
+ path: string;
+}
+
interface BundleHeader {
code: VerifiedByteRange;
metadata: VerifiedByteRange;
- source: VerifiedByteRange;
+ sources: VerifiedSourceRegion[];
+}
+
+/** Per-source-region metadata within the concatenated sources buffer. */
+interface SourceRegion {
+ path: string;
+ payloadStart: number;
+ payloadEnd: number;
+ sha256: string;
+}
+
+interface EncodedSources {
+ buffer: Buffer;
+ regions: SourceRegion[];
}
export function encodeBundle(input: BundleEncoderInput): Buffer {
const metadata = encodeMetadata(input);
- const source = encodeSource(input.entrypointSource);
+ const sources = encodeSources(input.entrypointSources);
const executable = encodeExecutable(input.executable);
- const header = encodeHeader({ metadata, source, executable });
+ const header = encodeHeader({ metadata, sources, executable });
- return Buffer.concat([header, metadata, source, executable]);
+ return Buffer.concat([header, metadata, sources.buffer, executable]);
}
-function encodeHeader(regions: { metadata: Buffer; source: Buffer; executable:
Buffer }): Buffer {
+function encodeHeader(regions: {
+ metadata: Buffer;
+ sources: EncodedSources;
+ executable: Buffer;
+}): Buffer {
// Each digest covers the payload only. The framing markers and newlines are
re-derived.
const metadataPayload = regions.metadata.subarray(
Buffer.byteLength(EMBEDDED_METADATA_PREFIX),
-1,
);
- const sourcePayload = regions.source.subarray(
- Buffer.byteLength(EMBEDDED_SOURCE_OPEN),
- -Buffer.byteLength(EMBEDDED_SOURCE_CLOSE),
- );
- const digests = {
- code: computeSha256(regions.executable),
- metadata: computeSha256(metadataPayload),
- source: computeSha256(sourcePayload),
- };
+ const metadataDigest = computeSha256(metadataPayload);
+ const codeDigest = computeSha256(regions.executable);
const zeroOffset = "0".repeat(OFFSET_HEX_WIDTH);
+ // Placeholder header with zeroed offsets and real digests, to measure its
+ // length without recursing. Every source path is present, so the array
+ // length is what it will be in the final header.
const placeholderHeader = renderHeader({
- code: { start: zeroOffset, end: zeroOffset, sha256: digests.code },
- metadata: { start: zeroOffset, end: zeroOffset, sha256: digests.metadata },
- source: { start: zeroOffset, end: zeroOffset, sha256: digests.source },
+ code: { start: zeroOffset, end: zeroOffset, sha256: codeDigest },
+ metadata: { start: zeroOffset, end: zeroOffset, sha256: metadataDigest },
+ sources: regions.sources.regions.map((region) => ({
+ path: region.path,
+ start: zeroOffset,
+ end: zeroOffset,
+ sha256: region.sha256,
+ })),
});
const metadataStart = placeholderHeader.length +
Buffer.byteLength(EMBEDDED_METADATA_PREFIX);
const metadataEnd = metadataStart + metadataPayload.length;
- const sourceStart =
- placeholderHeader.length + regions.metadata.length +
Buffer.byteLength(EMBEDDED_SOURCE_OPEN);
- const sourceEnd = sourceStart + sourcePayload.length;
- const codeStart = placeholderHeader.length + regions.metadata.length +
regions.source.length;
+ const sourcesBaseOffset = placeholderHeader.length + regions.metadata.length;
+ const codeStart = sourcesBaseOffset + regions.sources.buffer.length;
const codeEnd = codeStart + regions.executable.length;
const header = renderHeader({
code: {
start: formatOffset(codeStart),
end: formatOffset(codeEnd),
- sha256: digests.code,
+ sha256: codeDigest,
},
metadata: {
start: formatOffset(metadataStart),
end: formatOffset(metadataEnd),
- sha256: digests.metadata,
- },
- source: {
- start: formatOffset(sourceStart),
- end: formatOffset(sourceEnd),
- sha256: digests.source,
+ sha256: metadataDigest,
},
+ sources: regions.sources.regions.map((region) => ({
+ path: region.path,
+ start: formatOffset(sourcesBaseOffset + region.payloadStart),
+ end: formatOffset(sourcesBaseOffset + region.payloadEnd),
+ sha256: region.sha256,
+ })),
Review Comment:
Fixed in 709a42f1c30 — the header is now encoded as utf-8. A non-ASCII path
in `sources[i].path` round-trips correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
