Draft: Add auto-generated plugins to import data to Splitgraph from all supported data sources (Airbyte, Singer Taps, etc), and add export plugin to export from Splitgraph to Seafowl via Splitgraph GraphQL API

[milesrichardson]

WIP

Example (initialization part is pseudocode as it needs to be cleaned up)

// This part is pseudocode as it needs to be cleaned up (some junk needs to be passed as plugin args atm)
const { db } = makeSplitgraphHTTPContext({
    plugins: [
      new AirbyteGithubImportPlugin(),
    ],
    authenticatedCredential: {
      apiKey: credential.apiKey,
      apiSecret: credential.apiSecret,
      anonymous: false,
    },
  });

const res = await db.importData(
  "airbyte-github",
  {
    credentials: {
      credentials: {
        personal_access_token: GITHUB_PAT_SECRET,
      },
    },
    params: {
      repository: "splitgraph/seafowl",
      start_date: "2021-06-01T00:00:00Z",
    },
  },
  {
    namespace: namespace,
    repository: "madatdata-test-github-ingestion",
    tables: [
      {
        name: "stargazers",
        options: {
          airbyte_cursor_field: ["starred_at"],
          airbyte_primary_key_field: [],
        },
        schema: [],
      },
    ],
  }
);

type-checking/autocomplete works based on which plugin is specified in the first parameter:

Result: https://www.splitgraph.com/miles/madatdata-test-github-ingestion/latest/-/tables

[milesrichardson]

Some stuff in this PR, mainly around refactoring and adding features to import and export plugins:

There is now an import plugin for all 100+ data sources supported by Splitgraph (Airbyte connectors, Singer taps, etc.)

Plugins can be "installed" by including them in the list of plugins passed to the constructor of the database (e.g. new SeafowlDb({plugins: [...]}) or new SplitgraphDb({plugins: [...]}), but usually called indirectly through helper functions like makeSplitgraphHTTPContext):

const { db } = makeSplitgraphHTTPContext({
    plugins: [
      new AirbyteGithubImportPlugin(),
    ],
    authenticatedCredential: {
      apiKey: credential.apiKey,
      apiSecret: credential.apiSecret,
      anonymous: false,
    },
  });

Each plugin has a name, in this case airbyte-github, which is then specified as the first parameter to db.exportData or db.importData or db.pollDeferredTask (see below). The rest of the parameters to that function are defined by the function implementation of the plugin and are typechecked/auto-completed accordingly.

For the auto-generated plugins, credentials and params are auto-completed and typechecked via auto-generated types which are created from the JSONSchema returned by the Splitgraph API.

For example, import a GitHub repository into a Splitgraph repository using the airbyte-github plugin:

// (helper function that adds `new AirbyteGithubImportPlugin()` to the list of plugins)
const db = createRealDb();

// You need to be authenticated to Splitgraph to use this plugin
// Get the token, to get your username, so that you can import to a new repository in your namespace
const { username: namespace } = await fetchToken(db);

await db.importData(
  "airbyte-github",
  {
    credentials: {
      credentials: {
        personal_access_token: GITHUB_PAT_SECRET,
      },
    },
    params: {
      repository: "splitgraph/seafowl",
      start_date: "2021-06-01T00:00:00Z",
    },
  },
  {
    namespace: namespace,
    repository: "madatdata-test-github-ingestion",
    tables: [
      {
        name: "stargazers",
        options: {
          airbyte_cursor_field: ["starred_at"],
          airbyte_primary_key_field: [],
        },
        schema: [],
      },
    ],
  }
);

If the repository does not exist yet, then it will be created. Otherwise, a new image will be created with the imported data. This uses the StartExternalRepositoryLoad mutation from the Splitgraph GraphQL API.

Splitgraph import and export plugins can be "deferred" with { defer: true } parameter

By default, importData an exportData will create the import/export jobs via the Splitgraph API, and then wait for them to complete by polling the job status. However, this might not always be desirable. For example, perhaps you are submitting the job from the server side because it requires a secret (like your Splitgraph API_SECRET, or your credential for whatever upstream data source you're importing, like a personal access token for airbyte-github). In this case, you don't want the server side request to block until the ingestion job is complete, because it would likely timeout or hit limits of your Edge Function ™ vendor.

Now you can pass { defer: true } to the importData and exportData functions, and they will return immediately with a serialized task description containing a taskId (or multiple taskIds in the case of an export of table(s)/quer(y)(ies)/vdb(s) to Seafowl). Then, you can call pollDeferredTask with a parameter for taskId to check whether the job has completed yet. The idea is that you would have one Edge Function ™ to submit the ingestion job, and another Edge Function ™ to check whether a taskId has completed yet (effectively acting as a relatively dumb proxy to the Splitgraph API). Then the client side doesn't need any secrets, and it can poll the edge function with reasonable granularity.

Note: For Seafowl, we still have a SeafowlImportFilePlugin which "imports" a file to Seafowl by uploading it to the multi-part upload endpoint, but this plugin is not deferrable because it actually does the work of uploading the file directly to Seafowl. If you want a deferred export to Seafowl, you probably want to go through Splitgraph first (by uploading the file to Splitgraph, and then using Splitgraph to export to Seafowl; since the actual export happens on the Splitgraph backend, it's deferrable).

Example: Defer export of query result from Splitgraph to Seafowl

Execute a query on Splitgraph and export its result to a table in a Seafowl DB. This uses the Splitgraph API and can also export multiple tables, vdbs, or queries at once. A separate job with its own taskId is created and returned for each.

(Notice the { defer: true } in the last parameter)

// Export the result of this query from Splitgraph to a Seafowl instance
const queryToExport = `SELECT a as int_val, string_agg(random()::text, '') as text_val
FROM generate_series(1, 5) a, generate_series(1, 50) b
GROUP BY a ORDER BY a;`;

const res = await db.exportData(
  "export-to-seafowl",
  {
    queries: [
      {
        source: {
          query: queryToExport,
        },
        destination: {
          schema: "madatdata_testing",
          table: destTable,
        },
      },
    ],
  },
  {
    seafowlInstance: {
      selfHosted: {
        url: SEAFOWL_DEST_URL,
        dbname: SEAFOWL_DEST_DBNAME,
        secret: SEAFOWL_DEST_SECRET,
      },
    },
  },
  { defer: true }
);

Then, check if it's completed yet (in this example it hasn't) (mind the messy types):

const startedTask = await db.pollDeferredTask("export-to-seafowl", {
  taskId: taskId as string,
});

expect(startedTask.completed).toBe(false);

Example: Defer export of Splitgraph query to Parquet file

Execute a query on Splitgraph and export its result to a Parquet file at a publicly accessible, signed temporary URL that will expire in 24 hours:

const {
  taskId,
  response,
} = await db.exportData(
  "export-query-to-file",
  {
    query: `SELECT a as int_val, string_agg(random()::text, '') as text_val
FROM generate_series(1, 5) a, generate_series(1, 50) b
GROUP BY a ORDER BY a;`,
    vdbId: "ddn",
  },
  {
    format: "parquet",
    filename: "random-series",
  },
  { defer: true }
);

Check whether the export is complete:

const startedTask = await db.pollDeferredTask("export-query-to-file", {
  taskId: taskId as string,
});

expect(startedTask.completed).toBe(false);

Once it's completed, the URL of the exported file will be available (just like it would be available in the return value of exportData when called without { defer: true }):

const shouldBeCompletedTask = await db.pollDeferredTask(
  "export-query-to-file",
  { taskId: response.taskId }
);

const parquetFileURL = shouldBeCompletedTask.response?.output.url;

Example: Defer import of uploaded CSV to table on Splitgraph DDN

Upload a CSV to object storage (using a pre-signed upload URL from Splitgraph), then provide that URL to the Splitgraph API to "import" the CSV as a table on the Splitgraph DDN

(Each new export creates an image. You can see the results of the integration testing in my public repo at https://splitgraph.com/miles/dunno )

const { response, info, taskId } = await db.importData(
  "csv",
  { data: Buffer.from(`name;candies\r\nBob;5\r\nAlice;10`) },
  {
    tableName: `irrelevant-${randSuffix()}`,
    namespace,
    repository: "dunno",
    tableParams: {
      delimiter: ";",
    },
  },
  { defer: true }
);

Check whether it's complete - in this case you also need to provide a namespace and repository, but those parameters are typechecked (again, ignore the as string 👀):

const startedTask = await db.pollDeferredTask("csv", {
  taskId: taskId as string,
  namespace,
  repository: "dunno",
});

expect(startedTask.completed).toBe(false); // we just started it, so it's not completed yet :)