Fix/csv import existing table (#1573)

.changeset/quick-sheep-buy.md

@@ -0,0 +1,5 @@
+---
+'@xata.io/cli': patch
+---
+
+add ability to import new data to existing tables

packages/cli/src/commands/import/csv.test.ts

@@ -1,5 +1,68 @@
 import { describe, expect, test } from 'vitest';
 import { splitCommas } from './csv';
+import { compareSchemas } from '../../utils/compareSchema';
+import { Schemas } from '@xata.io/client';
+
+const defaultTableInfo = {
+  name: 'one',
+  xataCompatible: true,
+  comment: '',
+  primaryKey: ['id'],
+  uniqueConstraints: {},
+  checkConstraints: {},
+  foreignKeys: {},
+  indexes: {},
+  oid: ''
+};
+const sharedIdCol = {
+  id: {
+    name: 'id',
+    type: 'integer',
+    comment: '',
+    nullable: false,
+    unique: true,
+    default: null
+  }
+};
+const sourceSchemaDefault: Schemas.BranchSchema = {
+  name: 'main',
+  tables: {
+    one: {
+      ...defaultTableInfo,
+      columns: {
+        ...sharedIdCol,
+        colToDelete: {
+          name: 'colToDelete',
+          type: 'text',
+          comment: '',
+          nullable: false,
+          unique: false,
+          default: null
+        }
+      }
+    }
+  }
+};
+
+const targetSchemaDefault: Schemas.BranchSchema = {
+  name: 'main',
+  tables: {
+    one: {
+      ...defaultTableInfo,
+      columns: {
+        ...sharedIdCol,
+        newCol: {
+          name: 'newCol',
+          type: 'text',
+          comment: '',
+          nullable: false,
+          unique: false,
+          default: null
+        }
+      }
+    }
+  }
+};
 
 describe('splitCommas', () => {
   test('returns [] for falsy values', () => {
@@ -13,3 +76,127 @@ describe('splitCommas', () => {
     expect(splitCommas('a,b,c')).toEqual(['a', 'b', 'c']);
   });
 });
+
+describe('compare schemas', () => {
+  test('returns an empty array for identical schemas', () => {
+    const { edits } = compareSchemas({ source: sourceSchemaDefault, target: sourceSchemaDefault });
+    expect(edits).toEqual([]);
+  });
+  test('ignores internal columns on source', () => {
+    const { edits } = compareSchemas({
+      source: {
+        ...sourceSchemaDefault,
+        tables: {
+          ...sourceSchemaDefault.tables,
+          one: {
+            ...sourceSchemaDefault.tables.one,
+            columns: {
+              ...sourceSchemaDefault.tables.one.columns,
+              xata_id: {
+                name: 'xata_id',
+                type: 'text',
+                comment: '',
+                nullable: false,
+                unique: false,
+                default: null
+              }
+            }
+          }
+        }
+      },
+      target: targetSchemaDefault
+    });
+    expect(edits).toMatchInlineSnapshot(compareSnapshot);
+  });
+  test('ignores internal columns on target', () => {
+    const { edits } = compareSchemas({
+      source: sourceSchemaDefault,
+      target: {
+        ...targetSchemaDefault,
+        tables: {
+          ...targetSchemaDefault.tables,
+          one: {
+            ...targetSchemaDefault.tables.one,
+            columns: {
+              ...targetSchemaDefault.tables.one.columns,
+              xata_id: {
+                name: 'xata_id',
+                type: 'text',
+                comment: '',
+                nullable: false,
+                unique: false,
+                default: null
+              }
+            }
+          }
+        }
+      }
+    });
+    expect(edits).toMatchInlineSnapshot(compareSnapshot);
+  });
+  test('returns an array with create table if table does not already exist', () => {
+    const { edits } = compareSchemas({ source: {}, target: targetSchemaDefault });
+    expect(edits).toMatchInlineSnapshot(`
+      [
+        {
+          "create_table": {
+            "columns": [
+              {
+                "comment": "",
+                "default": undefined,
+                "name": "id",
+                "nullable": false,
+                "references": undefined,
+                "type": "integer",
+                "unique": true,
+              },
+              {
+                "comment": "",
+                "default": undefined,
+                "name": "newCol",
+                "nullable": false,
+                "references": undefined,
+                "type": "text",
+                "unique": false,
+              },
+            ],
+            "comment": "",
+            "name": "one",
+          },
+        },
+      ]
+    `);
+  });
+  test('returns an array with add_column for new columns', () => {
+    const { edits } = compareSchemas({ source: sourceSchemaDefault, target: targetSchemaDefault });
+    expect(edits).toMatchInlineSnapshot(compareSnapshot);
+  });
+  test('returns an array with drop_column for deleted columns', () => {
+    const { edits } = compareSchemas({ source: sourceSchemaDefault, target: targetSchemaDefault });
+    expect(edits).toMatchInlineSnapshot(compareSnapshot);
+  });
+});
+
+const compareSnapshot = `
+[
+  {
+    "add_column": {
+      "column": {
+        "comment": "",
+        "default": undefined,
+        "name": "newCol",
+        "nullable": false,
+        "references": undefined,
+        "type": "text",
+        "unique": false,
+      },
+      "table": "one",
+    },
+  },
+  {
+    "drop_column": {
+      "column": "colToDelete",
+      "table": "one",
+    },
+  },
+]`;

packages/cli/src/commands/import/csv.ts

@@ -8,11 +8,9 @@ import { enumFlag } from '../../utils/oclif.js';
 import {
   getBranchDetailsWithPgRoll,
   isBranchPgRollEnabled,
-  waitForMigrationToFinish,
-  xataColumnTypeToPgRollComment
+  waitForMigrationToFinish
 } from '../../migrations/pgroll.js';
-import { compareSchemas } from '../../utils/compareSchema.js';
-import { keyBy } from 'lodash';
+import { compareSchemas, inferOldSchemaToNew } from '../../utils/compareSchema.js';
 
 const ERROR_CONSOLE_LOG_LIMIT = 200;
 const ERROR_LOG_FILE = 'errors.log';
@@ -30,7 +28,7 @@ const bufferEncodings: BufferEncoding[] = [
   'hex'
 ];
 
-const INTERNAL_COLUMNS_PGROLL = ['xata_id', 'xata_createdat', 'xata_updatedat', 'xata_version'];
+export const INTERNAL_COLUMNS_PGROLL = ['xata_id', 'xata_createdat', 'xata_updatedat', 'xata_version'];
 
 export default class ImportCSV extends BaseCommand<typeof ImportCSV> {
   static description = 'Import a CSV file';
@@ -246,6 +244,7 @@ export default class ImportCSV extends BaseCommand<typeof ImportCSV> {
     const xata = await this.getXataClient();
     const { workspace, region, database, branch } = await this.parseDatabase();
     const { schema: existingSchema } = await getBranchDetailsWithPgRoll(xata, { workspace, region, database, branch });
+
     const newSchema = {
       tables: [
         ...existingSchema.tables.filter((t) => t.name !== table),
@@ -254,33 +253,12 @@ export default class ImportCSV extends BaseCommand<typeof ImportCSV> {
     };
 
     if (this.#pgrollEnabled) {
-      const { edits } = compareSchemas(
-        {},
-        {
-          tables: {
-            [table]: {
-              name: table,
-              xataCompatible: false,
-              columns: keyBy(
-                columns
-                  .filter((c) => !INTERNAL_COLUMNS_PGROLL.includes(c.name as any))
-                  .map((c) => {
-                    return {
-                      name: c.name,
-                      type: c.type,
-                      nullable: c.notNull !== false,
-                      default: c.defaultValue ?? null,
-                      unique: c.unique,
-                      comment: xataColumnTypeToPgRollComment(c)
-                    };
-                  }),
-                'name'
-              )
-            }
-          }
-        }
-      );
-
+      const sourceSchema = inferOldSchemaToNew({ schema: existingSchema, branchName: branch });
+      const targetSchema = inferOldSchemaToNew({ schema: newSchema, branchName: branch });
+      const { edits } = compareSchemas({
+        source: sourceSchema,
+        target: targetSchema
+      });
       if (edits.length > 0) {
         const destructiveOperations = edits
           .map((op) => {
@@ -324,6 +302,7 @@ export default class ImportCSV extends BaseCommand<typeof ImportCSV> {
           pathParams: { workspace, region, dbBranchName: `${database}:${branch}` },
           body: { operations: edits, adaptTables: true }
         });
+
         await waitForMigrationToFinish(xata.api, workspace, region, database, branch, jobID);
       }
     } else {