From 959d65f818bfb744ed1a0a079dc814af9a2743ac Mon Sep 17 00:00:00 2001 From: Tommy Healy Date: Tue, 30 Jun 2026 15:27:38 +0100 Subject: [PATCH] Add IMPORT_FORMAT support to Typescript SDK --- CHANGELOG.md | 4 ++ src/cli/commands/migrate.test.ts | 42 +++++++++++++++++++ src/generator/datasource.test.ts | 71 ++++++++++++++++++++++++++++++++ src/generator/datasource.ts | 4 ++ src/migrate/emit-ts.ts | 8 ++++ src/migrate/parse-datasource.ts | 12 +++++- src/migrate/types.ts | 2 + src/schema/datasource.ts | 4 ++ 8 files changed, 146 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ea4b791..85f0a2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ## [Unreleased] +### Added + +- S3 and GCS import data sources accept an optional `importFormat` (`csv`/`ndjson`/`parquet`), emitted as `IMPORT_FORMAT` in the generated `.datasource` and round-tripped by the datafile parser and migration emitter. Lets you ingest files whose extension does not imply the format (for example NDJSON delivered as `.log`), where the connector would otherwise fail with `Format not supported`. + ## [0.0.80] - 2026-06-24 ### Security diff --git a/src/cli/commands/migrate.test.ts b/src/cli/commands/migrate.test.ts index 0bef5b0..819c1b4 100644 --- a/src/cli/commands/migrate.test.ts +++ b/src/cli/commands/migrate.test.ts @@ -617,6 +617,48 @@ IMPORT_FROM_TIMESTAMP 2024-01-01T00:00:00Z expect(output).toContain('fromTimestamp: "2024-01-01T00:00:00Z"'); }); + it("round-trips IMPORT_FORMAT for s3 import datasource directives", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "tinybird-migrate-")); + tempDirs.push(tempDir); + + writeFile( + tempDir, + "s3sample.connection", + `TYPE s3 +S3_REGION "us-east-1" +S3_ARN "arn:aws:iam::123456789012:role/tinybird-s3-access" +` + ); + + writeFile( + tempDir, + "events_landing.datasource", + `SCHEMA > + timestamp DateTime, + session_id String + +ENGINE "MergeTree" +ENGINE_SORTING_KEY "timestamp" +IMPORT_CONNECTION_NAME s3sample +IMPORT_BUCKET_URI s3://my-bucket/events/*.log +IMPORT_FORMAT "ndjson" +` + ); + + const result = await runMigrate({ + cwd: tempDir, + patterns: ["."], + strict: true, + }); + + expect(result.success).toBe(true); + expect(result.errors).toHaveLength(0); + + const output = fs.readFileSync(result.outputPath, "utf-8"); + expect(output).toContain('bucketUri: "s3://my-bucket/events/*.log"'); + expect(output).toContain('importFormat: "ndjson"'); + }); + it("migrates gcs connection and import datasource directives", async () => { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "tinybird-migrate-")); tempDirs.push(tempDir); diff --git a/src/generator/datasource.test.ts b/src/generator/datasource.test.ts index 3a83664..8ce0ff0 100644 --- a/src/generator/datasource.test.ts +++ b/src/generator/datasource.test.ts @@ -598,6 +598,54 @@ describe('Datasource Generator', () => { expect(result.content).toContain('IMPORT_SCHEDULE @auto'); expect(result.content).toContain('IMPORT_FROM_TIMESTAMP 2024-01-01T00:00:00Z'); }); + + it('emits IMPORT_FORMAT when importFormat is set', () => { + const s3Conn = defineS3Connection('my_s3', { + region: 'us-east-1', + arn: 'arn:aws:iam::123456789012:role/tinybird-s3-access', + }); + + const ds = defineDatasource('s3_events', { + schema: { + timestamp: t.dateTime(), + event: t.string(), + }, + engine: engine.mergeTree({ sortingKey: ['timestamp'] }), + s3: { + connection: s3Conn, + bucketUri: 's3://my-bucket/events/*.log', + importFormat: 'ndjson', + }, + }); + + const result = generateDatasource(ds); + + expect(result.content).toContain('IMPORT_BUCKET_URI s3://my-bucket/events/*.log'); + expect(result.content).toContain('IMPORT_FORMAT "ndjson"'); + }); + + it('omits IMPORT_FORMAT when importFormat is unset', () => { + const s3Conn = defineS3Connection('my_s3', { + region: 'us-east-1', + arn: 'arn:aws:iam::123456789012:role/tinybird-s3-access', + }); + + const ds = defineDatasource('s3_events', { + schema: { + timestamp: t.dateTime(), + event: t.string(), + }, + engine: engine.mergeTree({ sortingKey: ['timestamp'] }), + s3: { + connection: s3Conn, + bucketUri: 's3://my-bucket/events/*.ndjson', + }, + }); + + const result = generateDatasource(ds); + + expect(result.content).not.toContain('IMPORT_FORMAT'); + }); }); describe('GCS configuration', () => { @@ -627,6 +675,29 @@ describe('Datasource Generator', () => { expect(result.content).toContain('IMPORT_SCHEDULE @auto'); expect(result.content).toContain('IMPORT_FROM_TIMESTAMP 2024-01-01T00:00:00Z'); }); + + it('emits IMPORT_FORMAT when importFormat is set', () => { + const gcsConn = defineGCSConnection('my_gcs', { + serviceAccountCredentialsJson: '{{ tb_secret("GCS_SERVICE_ACCOUNT_CREDENTIALS_JSON") }}', + }); + + const ds = defineDatasource('gcs_events', { + schema: { + timestamp: t.dateTime(), + event: t.string(), + }, + engine: engine.mergeTree({ sortingKey: ['timestamp'] }), + gcs: { + connection: gcsConn, + bucketUri: 'gs://my-bucket/events/*.log', + importFormat: 'ndjson', + }, + }); + + const result = generateDatasource(ds); + + expect(result.content).toContain('IMPORT_FORMAT "ndjson"'); + }); }); describe('DynamoDB configuration', () => { diff --git a/src/generator/datasource.ts b/src/generator/datasource.ts index ba19504..6a653dc 100644 --- a/src/generator/datasource.ts +++ b/src/generator/datasource.ts @@ -197,6 +197,10 @@ function generateImportConfig(importConfig: S3Config | GCSConfig): string { parts.push(`IMPORT_FROM_TIMESTAMP ${importConfig.fromTimestamp}`); } + if (importConfig.importFormat) { + parts.push(`IMPORT_FORMAT "${importConfig.importFormat}"`); + } + return parts.join("\n"); } diff --git a/src/migrate/emit-ts.ts b/src/migrate/emit-ts.ts index 473eb10..273e47c 100644 --- a/src/migrate/emit-ts.ts +++ b/src/migrate/emit-ts.ts @@ -93,11 +93,13 @@ function hasSecretTemplate(resources: ParsedResource[]): boolean { values.push(resource.s3.bucketUri); if (resource.s3.schedule) values.push(resource.s3.schedule); if (resource.s3.fromTimestamp) values.push(resource.s3.fromTimestamp); + if (resource.s3.importFormat) values.push(resource.s3.importFormat); } if (resource.gcs) { values.push(resource.gcs.bucketUri); if (resource.gcs.schedule) values.push(resource.gcs.schedule); if (resource.gcs.fromTimestamp) values.push(resource.gcs.fromTimestamp); + if (resource.gcs.importFormat) values.push(resource.gcs.importFormat); } if (resource.dynamodb) { values.push(resource.dynamodb.tableArn); @@ -379,6 +381,9 @@ function emitDatasource(ds: DatasourceModel): string { if (ds.s3.fromTimestamp) { lines.push(` fromTimestamp: ${emitStringOrSecret(ds.s3.fromTimestamp)},`); } + if (ds.s3.importFormat) { + lines.push(` importFormat: ${emitStringOrSecret(ds.s3.importFormat)},`); + } lines.push(" },"); } @@ -393,6 +398,9 @@ function emitDatasource(ds: DatasourceModel): string { if (ds.gcs.fromTimestamp) { lines.push(` fromTimestamp: ${emitStringOrSecret(ds.gcs.fromTimestamp)},`); } + if (ds.gcs.importFormat) { + lines.push(` importFormat: ${emitStringOrSecret(ds.gcs.importFormat)},`); + } lines.push(" },"); } diff --git a/src/migrate/parse-datasource.ts b/src/migrate/parse-datasource.ts index fb653d2..9b79466 100644 --- a/src/migrate/parse-datasource.ts +++ b/src/migrate/parse-datasource.ts @@ -39,6 +39,7 @@ const DATASOURCE_DIRECTIVES = new Set([ "IMPORT_BUCKET_URI", "IMPORT_SCHEDULE", "IMPORT_FROM_TIMESTAMP", + "IMPORT_FORMAT", "IMPORT_TABLE_ARN", "IMPORT_EXPORT_BUCKET", "TOKEN", @@ -323,6 +324,7 @@ export function parseDatasourceFile(resource: ResourceFile): DatasourceModel { let importBucketUri: string | undefined; let importSchedule: string | undefined; let importFromTimestamp: string | undefined; + let importFormat: string | undefined; let importTableArn: string | undefined; let importExportBucket: string | undefined; @@ -502,6 +504,9 @@ export function parseDatasourceFile(resource: ResourceFile): DatasourceModel { case "IMPORT_FROM_TIMESTAMP": importFromTimestamp = parseQuotedValue(value); break; + case "IMPORT_FORMAT": + importFormat = parseQuotedValue(value); + break; case "IMPORT_TABLE_ARN": importTableArn = parseQuotedValue(value); break; @@ -606,12 +611,17 @@ export function parseDatasourceFile(resource: ResourceFile): DatasourceModel { const s3 = !isDynamoDB && - (importConnectionName || importBucketUri || importSchedule || importFromTimestamp) + (importConnectionName || + importBucketUri || + importSchedule || + importFromTimestamp || + importFormat) ? { connectionName: importConnectionName ?? "", bucketUri: importBucketUri ?? "", schedule: importSchedule, fromTimestamp: importFromTimestamp, + importFormat, } : undefined; diff --git a/src/migrate/types.ts b/src/migrate/types.ts index 05b48e1..d63a746 100644 --- a/src/migrate/types.ts +++ b/src/migrate/types.ts @@ -50,6 +50,7 @@ export interface DatasourceS3Model { bucketUri: string; schedule?: string; fromTimestamp?: string; + importFormat?: string; } export interface DatasourceGCSModel { @@ -57,6 +58,7 @@ export interface DatasourceGCSModel { bucketUri: string; schedule?: string; fromTimestamp?: string; + importFormat?: string; } export interface DatasourceDynamoDBModel { diff --git a/src/schema/datasource.ts b/src/schema/datasource.ts index c8d23f4..e0cb0f3 100644 --- a/src/schema/datasource.ts +++ b/src/schema/datasource.ts @@ -87,6 +87,8 @@ export interface S3Config { schedule?: string; /** Incremental import lower bound timestamp expression */ fromTimestamp?: string; + /** Explicit import format (`csv`/`ndjson`/`parquet`) when the file extension does not imply it */ + importFormat?: string; } /** @@ -101,6 +103,8 @@ export interface GCSConfig { schedule?: string; /** Incremental import lower bound timestamp expression */ fromTimestamp?: string; + /** Explicit import format (`csv`/`ndjson`/`parquet`) when the file extension does not imply it */ + importFormat?: string; } /**