PROTM-1309: Add IMPORT_FORMAT support to Typescript SDK

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

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);

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', () => {

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");
 }
 

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("  },");
   }
 

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;
 

src/migrate/types.ts

@@ -50,13 +50,15 @@ export interface DatasourceS3Model {
   bucketUri: string;
   schedule?: string;
   fromTimestamp?: string;
+  importFormat?: string;
 }
 
 export interface DatasourceGCSModel {
   connectionName: string;
   bucketUri: string;
   schedule?: string;
   fromTimestamp?: string;
+  importFormat?: string;
 }
 
 export interface DatasourceDynamoDBModel {

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;
 }
 
 /**