diff --git a/superset-frontend/packages/superset-ui-core/src/query/types/Query.ts b/superset-frontend/packages/superset-ui-core/src/query/types/Query.ts
index 4d5ccaf6f09b1..b6b1fd3a638cc 100644
--- a/superset-frontend/packages/superset-ui-core/src/query/types/Query.ts
+++ b/superset-frontend/packages/superset-ui-core/src/query/types/Query.ts
@@ -316,6 +316,7 @@ export type Query = {
   link?: string;
   progress: number;
   resultsKey: string | null;
+  catalog?: string | null;
   schema?: string;
   sql: string;
   sqlEditorId: string;
diff --git a/superset-frontend/packages/superset-ui-core/src/ui-overrides/types.ts b/superset-frontend/packages/superset-ui-core/src/ui-overrides/types.ts
index f8f10982bd1f9..736a57da900a3 100644
--- a/superset-frontend/packages/superset-ui-core/src/ui-overrides/types.ts
+++ b/superset-frontend/packages/superset-ui-core/src/ui-overrides/types.ts
@@ -168,6 +168,7 @@ export interface SubMenuProps {
 export interface CustomAutoCompleteArgs {
   queryEditorId: string;
   dbId?: string | number;
+  catalog?: string | null;
   schema?: string;
 }
 
diff --git a/superset-frontend/src/SqlLab/actions/sqlLab.js b/superset-frontend/src/SqlLab/actions/sqlLab.js
index 6befa17ac18d4..a153c4eb450a1 100644
--- a/superset-frontend/src/SqlLab/actions/sqlLab.js
+++ b/superset-frontend/src/SqlLab/actions/sqlLab.js
@@ -55,6 +55,7 @@ export const REMOVE_QUERY = 'REMOVE_QUERY';
 export const EXPAND_TABLE = 'EXPAND_TABLE';
 export const COLLAPSE_TABLE = 'COLLAPSE_TABLE';
 export const QUERY_EDITOR_SETDB = 'QUERY_EDITOR_SETDB';
+export const QUERY_EDITOR_SET_CATALOG = 'QUERY_EDITOR_SET_CATALOG';
 export const QUERY_EDITOR_SET_SCHEMA = 'QUERY_EDITOR_SET_SCHEMA';
 export const QUERY_EDITOR_SET_TITLE = 'QUERY_EDITOR_SET_TITLE';
 export const QUERY_EDITOR_SET_AUTORUN = 'QUERY_EDITOR_SET_AUTORUN';
@@ -326,6 +327,7 @@ export function runQuery(query) {
       database_id: query.dbId,
       json: true,
       runAsync: query.runAsync,
+      catalog: query.catalog,
       schema: query.schema,
       sql: query.sql,
       sql_editor_id: query.sqlEditorId,
@@ -381,6 +383,7 @@ export function runQueryFromSqlEditor(
       sql: qe.selectedText || qe.sql,
       sqlEditorId: qe.id,
       tab: qe.name,
+      catalog: qe.catalog,
       schema: qe.schema,
       tempTable,
       templateParams: qe.templateParams,
@@ -556,7 +559,7 @@ export function addNewQueryEditor() {
     );
     const dbIds = Object.values(databases).map(database => database.id);
     const firstDbId = dbIds.length > 0 ? Math.min(...dbIds) : undefined;
-    const { dbId, schema, queryLimit, autorun } = {
+    const { dbId, catalog, schema, queryLimit, autorun } = {
       ...queryEditors[0],
       ...activeQueryEditor,
       ...(unsavedQueryEditor.id === activeQueryEditor?.id &&
@@ -578,6 +581,7 @@ export function addNewQueryEditor() {
     return dispatch(
       addQueryEditor({
         dbId: dbId || defaultDbId || firstDbId,
+        catalog: catalog ?? null,
         schema: schema ?? null,
         autorun: autorun ?? false,
         sql: `${warning}SELECT ...`,
@@ -600,6 +604,7 @@ export function cloneQueryToNewTab(query, autorun) {
     const queryEditor = {
       name: t('Copy of %s', sourceQueryEditor.name),
       dbId: query.dbId ? query.dbId : null,
+      catalog: query.catalog ? query.catalog : null,
       schema: query.schema ? query.schema : null,
       autorun,
       sql: query.sql,
@@ -656,6 +661,7 @@ export function setTables(tableSchemas) {
       return {
         dbId: tableSchema.database_id,
         queryEditorId: tableSchema.tab_state_id.toString(),
+        catalog: tableSchema.catalog,
         schema: tableSchema.schema,
         name: tableSchema.table,
         expanded: tableSchema.expanded,
@@ -694,6 +700,7 @@ export function switchQueryEditor(queryEditor, displayLimit) {
             autorun: json.autorun,
             dbId: json.database_id,
             templateParams: json.template_params,
+            catalog: json.catalog,
             schema: json.schema,
             queryLimit: json.query_limit,
             remoteId: json.saved_query?.id,
@@ -797,6 +804,14 @@ export function queryEditorSetDb(queryEditor, dbId) {
   return { type: QUERY_EDITOR_SETDB, queryEditor, dbId };
 }
 
+export function queryEditorSetCatalog(queryEditor, catalog) {
+  return {
+    type: QUERY_EDITOR_SET_CATALOG,
+    queryEditor: queryEditor || {},
+    catalog,
+  };
+}
+
 export function queryEditorSetSchema(queryEditor, schema) {
   return {
     type: QUERY_EDITOR_SET_SCHEMA,
@@ -954,12 +969,13 @@ export function mergeTable(table, query, prepend) {
   return { type: MERGE_TABLE, table, query, prepend };
 }
 
-export function addTable(queryEditor, tableName, schemaName) {
+export function addTable(queryEditor, tableName, catalogName, schemaName) {
   return function (dispatch, getState) {
     const query = getUpToDateQuery(getState(), queryEditor, queryEditor.id);
     const table = {
       dbId: query.dbId,
       queryEditorId: query.id,
+      catalog: catalogName,
       schema: schemaName,
       name: tableName,
     };
@@ -983,12 +999,14 @@ export function runTablePreviewQuery(newTable) {
       sqlLab: { databases },
     } = getState();
     const database = databases[newTable.dbId];
-    const { dbId } = newTable;
+    const { dbId, catalog, schema } = newTable;
 
     if (database && !database.disable_data_preview) {
       const dataPreviewQuery = {
         id: shortid.generate(),
         dbId,
+        catalog,
+        schema,
         sql: newTable.selectStar,
         tableName: newTable.name,
         sqlEditorId: null,
@@ -1003,6 +1021,7 @@ export function runTablePreviewQuery(newTable) {
             {
               id: newTable.id,
               dbId: newTable.dbId,
+              catalog: newTable.catalog,
               schema: newTable.schema,
               name: newTable.name,
               queryEditorId: newTable.queryEditorId,
@@ -1180,6 +1199,7 @@ export function popStoredQuery(urlId) {
           addQueryEditor({
             name: json.name ? json.name : t('Shared query'),
             dbId: json.dbId ? parseInt(json.dbId, 10) : null,
+            catalog: json.catalog ? json.catalog : null,
             schema: json.schema ? json.schema : null,
             autorun: json.autorun ? json.autorun : false,
             sql: json.sql ? json.sql : 'SELECT ...',
@@ -1215,6 +1235,7 @@ export function popQuery(queryId) {
         const queryData = json.result;
         const queryEditorProps = {
           dbId: queryData.database.id,
+          catalog: queryData.catalog,
           schema: queryData.schema,
           sql: queryData.sql,
           name: t('Copy of %s', queryData.tab_name),
@@ -1268,12 +1289,13 @@ export function createDatasourceFailed(err) {
 export function createDatasource(vizOptions) {
   return dispatch => {
     dispatch(createDatasourceStarted());
-    const { dbId, schema, datasourceName, sql } = vizOptions;
+    const { dbId, catalog, schema, datasourceName, sql } = vizOptions;
     return SupersetClient.post({
       endpoint: '/api/v1/dataset/',
       headers: { 'Content-Type': 'application/json' },
       body: JSON.stringify({
         database: dbId,
+        catalog,
         schema,
         sql,
         table_name: datasourceName,
diff --git a/superset-frontend/src/SqlLab/actions/sqlLab.test.js b/superset-frontend/src/SqlLab/actions/sqlLab.test.js
index 871b3ff6f6b4f..d6b70bf6a0b40 100644
--- a/superset-frontend/src/SqlLab/actions/sqlLab.test.js
+++ b/superset-frontend/src/SqlLab/actions/sqlLab.test.js
@@ -419,6 +419,7 @@ describe('async actions', () => {
           queryEditor: {
             name: 'Copy of Dummy query editor',
             dbId: 1,
+            catalog: query.catalog,
             schema: query.schema,
             autorun: true,
             sql: 'SELECT * FROM something',
@@ -481,6 +482,7 @@ describe('async actions', () => {
               sql: expect.stringContaining('SELECT ...'),
               name: `Untitled Query 7`,
               dbId: defaultQueryEditor.dbId,
+              catalog: defaultQueryEditor.catalog,
               schema: defaultQueryEditor.schema,
               autorun: false,
               queryLimit:
@@ -607,6 +609,24 @@ describe('async actions', () => {
       });
     });
 
+    describe('queryEditorSetCatalog', () => {
+      it('updates the tab state in the backend', () => {
+        expect.assertions(1);
+
+        const catalog = 'public';
+        const store = mockStore({});
+        const expectedActions = [
+          {
+            type: actions.QUERY_EDITOR_SET_CATALOG,
+            queryEditor,
+            catalog,
+          },
+        ];
+        store.dispatch(actions.queryEditorSetCatalog(queryEditor, catalog));
+        expect(store.getActions()).toEqual(expectedActions);
+      });
+    });
+
     describe('queryEditorSetSchema', () => {
       it('updates the tab state in the backend', () => {
         expect.assertions(1);
@@ -747,6 +767,7 @@ describe('async actions', () => {
     describe('addTable', () => {
       it('dispatches table state from unsaved change', () => {
         const tableName = 'table';
+        const catalogName = null;
         const schemaName = 'schema';
         const expectedDbId = 473892;
         const store = mockStore({
@@ -759,12 +780,18 @@ describe('async actions', () => {
             },
           },
         });
-        const request = actions.addTable(query, tableName, schemaName);
+        const request = actions.addTable(
+          query,
+          tableName,
+          catalogName,
+          schemaName,
+        );
         request(store.dispatch, store.getState);
         expect(store.getActions()[0]).toEqual(
           expect.objectContaining({
             table: expect.objectContaining({
               name: tableName,
+              catalog: catalogName,
               schema: schemaName,
               dbId: expectedDbId,
             }),
@@ -811,6 +838,7 @@ describe('async actions', () => {
         });
 
         const tableName = 'table';
+        const catalogName = null;
         const schemaName = 'schema';
         const store = mockStore({
           ...initialState,
@@ -829,6 +857,7 @@ describe('async actions', () => {
         const request = actions.runTablePreviewQuery({
           dbId: 1,
           name: tableName,
+          catalog: catalogName,
           schema: schemaName,
         });
         return request(store.dispatch, store.getState).then(() => {
diff --git a/superset-frontend/src/SqlLab/components/AceEditorWrapper/index.tsx b/superset-frontend/src/SqlLab/components/AceEditorWrapper/index.tsx
index 354d6b1c8fb81..219a1f9bbaad2 100644
--- a/superset-frontend/src/SqlLab/components/AceEditorWrapper/index.tsx
+++ b/superset-frontend/src/SqlLab/components/AceEditorWrapper/index.tsx
@@ -74,6 +74,7 @@ const AceEditorWrapper = ({
     'id',
     'dbId',
     'sql',
+    'catalog',
     'schema',
     'templateParams',
     'cursorPosition',
@@ -161,6 +162,7 @@ const AceEditorWrapper = ({
 
   const { data: annotations } = useAnnotations({
     dbId: queryEditor.dbId,
+    catalog: queryEditor.catalog,
     schema: queryEditor.schema,
     sql: currentSql,
     templateParams: queryEditor.templateParams,
@@ -170,6 +172,7 @@ const AceEditorWrapper = ({
     {
       queryEditorId,
       dbId: queryEditor.dbId,
+      catalog: queryEditor.catalog,
       schema: queryEditor.schema,
     },
     !autocomplete,
diff --git a/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.test.ts b/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.test.ts
index adee7d681815b..193e715fb220f 100644
--- a/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.test.ts
+++ b/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.test.ts
@@ -189,7 +189,12 @@ test('returns column keywords among selected tables', async () => {
     storeWithSqlLab.dispatch(
       tableApiUtil.upsertQueryData(
         'tableMetadata',
-        { dbId: expectDbId, schema: expectSchema, table: expectTable },
+        {
+          dbId: expectDbId,
+          catalog: null,
+          schema: expectSchema,
+          table: expectTable,
+        },
         {
           name: expectTable,
           columns: [
@@ -205,7 +210,12 @@ test('returns column keywords among selected tables', async () => {
     storeWithSqlLab.dispatch(
       tableApiUtil.upsertQueryData(
         'tableMetadata',
-        { dbId: expectDbId, schema: expectSchema, table: unexpectedTable },
+        {
+          dbId: expectDbId,
+          catalog: null,
+          schema: expectSchema,
+          table: unexpectedTable,
+        },
         {
           name: unexpectedTable,
           columns: [
@@ -227,6 +237,7 @@ test('returns column keywords among selected tables', async () => {
       useKeywords({
         queryEditorId: expectQueryEditorId,
         dbId: expectDbId,
+        catalog: null,
         schema: expectSchema,
       }),
     {
diff --git a/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.ts b/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.ts
index 09a50352600de..2eac21ff390b0 100644
--- a/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.ts
+++ b/superset-frontend/src/SqlLab/components/AceEditorWrapper/useKeywords.ts
@@ -42,6 +42,7 @@ import { SqlLabRootState } from 'src/SqlLab/types';
 type Params = {
   queryEditorId: string | number;
   dbId?: string | number;
+  catalog?: string | null;
   schema?: string;
 };
 
@@ -58,7 +59,7 @@ const getHelperText = (value: string) =>
 const extensionsRegistry = getExtensionsRegistry();
 
 export function useKeywords(
-  { queryEditorId, dbId, schema }: Params,
+  { queryEditorId, dbId, catalog, schema }: Params,
   skip = false,
 ) {
   const useCustomKeywords = extensionsRegistry.get(
@@ -68,6 +69,7 @@ export function useKeywords(
   const customKeywords = useCustomKeywords?.({
     queryEditorId: String(queryEditorId),
     dbId,
+    catalog,
     schema,
   });
   const dispatch = useDispatch();
@@ -78,6 +80,7 @@ export function useKeywords(
   const { data: schemaOptions } = useSchemasQueryState(
     {
       dbId,
+      catalog: catalog || undefined,
       forceRefresh: false,
     },
     { skip: skipFetch || !dbId },
@@ -85,6 +88,7 @@ export function useKeywords(
   const { data: tableData } = useTablesQueryState(
     {
       dbId,
+      catalog,
       schema,
       forceRefresh: false,
     },
@@ -125,6 +129,7 @@ export function useKeywords(
           dbId && schema
             ? {
                 dbId,
+                catalog,
                 schema,
                 table,
               }
@@ -137,7 +142,7 @@ export function useKeywords(
         });
     });
     return [...columns];
-  }, [dbId, schema, apiState, tablesForColumnMetadata]);
+  }, [dbId, catalog, schema, apiState, tablesForColumnMetadata]);
 
   const insertMatch = useEffectEvent((editor: Editor, data: any) => {
     if (data.meta === 'table') {
diff --git a/superset-frontend/src/SqlLab/components/SaveDatasetModal/SaveDatasetModal.test.tsx b/superset-frontend/src/SqlLab/components/SaveDatasetModal/SaveDatasetModal.test.tsx
index 8568bf20809e5..c3d1658f3a4ef 100644
--- a/superset-frontend/src/SqlLab/components/SaveDatasetModal/SaveDatasetModal.test.tsx
+++ b/superset-frontend/src/SqlLab/components/SaveDatasetModal/SaveDatasetModal.test.tsx
@@ -210,6 +210,38 @@ describe('SaveDatasetModal', () => {
     expect(createDatasource).toHaveBeenCalledWith({
       datasourceName: 'my dataset',
       dbId: 1,
+      catalog: null,
+      schema: 'main',
+      sql: 'SELECT *',
+      templateParams: undefined,
+    });
+  });
+
+  it('sends the catalog when creating the dataset', async () => {
+    const dummyDispatch = jest.fn().mockResolvedValue({});
+    useDispatchMock.mockReturnValue(dummyDispatch);
+    useSelectorMock.mockReturnValue({ ...user });
+
+    render(
+      <SaveDatasetModal
+        {...mockedProps}
+        datasource={{ ...mockedProps.datasource, catalog: 'public' }}
+      />,
+      { useRedux: true },
+    );
+
+    const inputFieldText = screen.getByDisplayValue(/unimportant/i);
+    fireEvent.change(inputFieldText, { target: { value: 'my dataset' } });
+
+    const saveConfirmationBtn = screen.getByRole('button', {
+      name: /save/i,
+    });
+    userEvent.click(saveConfirmationBtn);
+
+    expect(createDatasource).toHaveBeenCalledWith({
+      datasourceName: 'my dataset',
+      dbId: 1,
+      catalog: 'public',
       schema: 'main',
       sql: 'SELECT *',
       templateParams: undefined,
diff --git a/superset-frontend/src/SqlLab/components/SaveDatasetModal/index.tsx b/superset-frontend/src/SqlLab/components/SaveDatasetModal/index.tsx
index 885265fe8d2df..011d4a7d21189 100644
--- a/superset-frontend/src/SqlLab/components/SaveDatasetModal/index.tsx
+++ b/superset-frontend/src/SqlLab/components/SaveDatasetModal/index.tsx
@@ -77,6 +77,7 @@ export interface ISaveableDatasource {
   dbId: number;
   sql: string;
   templateParams?: string | object | null;
+  catalog?: string | null;
   schema?: string | null;
   database?: Database;
 }
@@ -292,6 +293,7 @@ export const SaveDatasetModal = ({
       createDatasource({
         sql: datasource.sql,
         dbId: datasource.dbId || datasource?.database?.id,
+        catalog: datasource?.catalog,
         schema: datasource?.schema,
         templateParams,
         datasourceName: datasetName,
diff --git a/superset-frontend/src/SqlLab/components/SaveQuery/SaveQuery.test.tsx b/superset-frontend/src/SqlLab/components/SaveQuery/SaveQuery.test.tsx
index 54b81df96013d..3d0413f62c72d 100644
--- a/superset-frontend/src/SqlLab/components/SaveQuery/SaveQuery.test.tsx
+++ b/superset-frontend/src/SqlLab/components/SaveQuery/SaveQuery.test.tsx
@@ -42,6 +42,7 @@ const mockState = {
       {
         id: mockedProps.queryEditorId,
         dbId: 1,
+        catalog: null,
         schema: 'main',
         sql: 'SELECT * FROM t',
       },
diff --git a/superset-frontend/src/SqlLab/components/SaveQuery/index.tsx b/superset-frontend/src/SqlLab/components/SaveQuery/index.tsx
index a7ac8b1b2a896..2cb50b0c066cb 100644
--- a/superset-frontend/src/SqlLab/components/SaveQuery/index.tsx
+++ b/superset-frontend/src/SqlLab/components/SaveQuery/index.tsx
@@ -48,7 +48,7 @@ export type QueryPayload = {
   description?: string;
   id?: string;
   remoteId?: number;
-} & Pick<QueryEditor, 'dbId' | 'schema' | 'sql'>;
+} & Pick<QueryEditor, 'dbId' | 'catalog' | 'schema' | 'sql'>;
 
 const Styles = styled.span`
   span[role='img'] {
@@ -78,6 +78,7 @@ const SaveQuery = ({
     'dbId',
     'latestQueryId',
     'queryLimit',
+    'catalog',
     'schema',
     'selectedText',
     'sql',
@@ -115,6 +116,7 @@ const SaveQuery = ({
     description,
     dbId: query.dbId ?? 0,
     sql: query.sql,
+    catalog: query.catalog,
     schema: query.schema,
     templateParams: query.templateParams,
     remoteId: query?.remoteId || undefined,
diff --git a/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/SqlEditorLeftBar.test.tsx b/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/SqlEditorLeftBar.test.tsx
index b5003b16f7b47..27d6c44d016b3 100644
--- a/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/SqlEditorLeftBar.test.tsx
+++ b/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/SqlEditorLeftBar.test.tsx
@@ -44,6 +44,10 @@ const mockedProps = {
 
 beforeEach(() => {
   fetchMock.get('glob:*/api/v1/database/?*', { result: [] });
+  fetchMock.get('glob:*/api/v1/database/*/catalogs/?*', {
+    count: 0,
+    result: [],
+  });
   fetchMock.get('glob:*/api/v1/database/*/schemas/?*', {
     count: 2,
     result: ['main', 'new_schema'],
@@ -103,11 +107,14 @@ test('renders a TableElement', async () => {
 });
 
 test('table should be visible when expanded is true', async () => {
-  const { container, getByText, getByRole, queryAllByText } =
-    await renderAndWait(mockedProps, undefined, {
+  const { container, getByText, getByRole } = await renderAndWait(
+    mockedProps,
+    undefined,
+    {
       ...initialState,
       sqlLab: { ...initialState.sqlLab, tables: [table] },
-    });
+    },
+  );
 
   const dbSelect = getByRole('combobox', {
     name: 'Select database or type to search databases',
@@ -115,14 +122,56 @@ test('table should be visible when expanded is true', async () => {
   const schemaSelect = getByRole('combobox', {
     name: 'Select schema or type to search schemas',
   });
-  const dropdown = getByText(/Table/i);
-  const abUser = queryAllByText(/ab_user/i);
+  const dropdown = getByText(/Select table/i);
+  const abUser = getByText(/ab_user/i);
+
+  expect(getByText(/Database/i)).toBeInTheDocument();
+  expect(dbSelect).toBeInTheDocument();
+  expect(schemaSelect).toBeInTheDocument();
+  expect(dropdown).toBeInTheDocument();
+  expect(abUser).toBeInTheDocument();
+  expect(
+    container.querySelector('.ant-collapse-content-active'),
+  ).toBeInTheDocument();
+  table.columns.forEach(({ name }) => {
+    expect(getByText(name)).toBeInTheDocument();
+  });
+});
+
+test('catalog selector should be visible when enabled in the database', async () => {
+  const { container, getByText, getByRole } = await renderAndWait(
+    {
+      ...mockedProps,
+      database: {
+        ...mockedProps.database,
+        allow_multi_catalog: true,
+      },
+    },
+    undefined,
+    {
+      ...initialState,
+      sqlLab: { ...initialState.sqlLab, tables: [table] },
+    },
+  );
+
+  const dbSelect = getByRole('combobox', {
+    name: 'Select database or type to search databases',
+  });
+  const catalogSelect = getByRole('combobox', {
+    name: 'Select catalog or type to search catalogs',
+  });
+  const schemaSelect = getByRole('combobox', {
+    name: 'Select schema or type to search schemas',
+  });
+  const dropdown = getByText(/Select table/i);
+  const abUser = getByText(/ab_user/i);
 
   expect(getByText(/Database/i)).toBeInTheDocument();
   expect(dbSelect).toBeInTheDocument();
+  expect(catalogSelect).toBeInTheDocument();
   expect(schemaSelect).toBeInTheDocument();
   expect(dropdown).toBeInTheDocument();
-  expect(abUser).toHaveLength(2);
+  expect(abUser).toBeInTheDocument();
   expect(
     container.querySelector('.ant-collapse-content-active'),
   ).toBeInTheDocument();
diff --git a/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/index.tsx b/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/index.tsx
index 15a17356268e3..1eee80f485a00 100644
--- a/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/index.tsx
+++ b/superset-frontend/src/SqlLab/components/SqlEditorLeftBar/index.tsx
@@ -34,6 +34,7 @@ import {
   removeTables,
   collapseTable,
   expandTable,
+  queryEditorSetCatalog,
   queryEditorSetSchema,
   setDatabases,
   addDangerToast,
@@ -115,13 +116,17 @@ const SqlEditorLeftBar = ({
     shallowEqual,
   );
   const dispatch = useDispatch();
-  const queryEditor = useQueryEditor(queryEditorId, ['dbId', 'schema']);
+  const queryEditor = useQueryEditor(queryEditorId, [
+    'dbId',
+    'catalog',
+    'schema',
+  ]);
 
   const [emptyResultsWithSearch, setEmptyResultsWithSearch] = useState(false);
   const [userSelectedDb, setUserSelected] = useState<DatabaseObject | null>(
     null,
   );
-  const { schema } = queryEditor;
+  const { catalog, schema } = queryEditor;
 
   useEffect(() => {
     const bool = querystring.parse(window.location.search).db;
@@ -138,9 +143,9 @@ const SqlEditorLeftBar = ({
     }
   }, [database]);
 
-  const onEmptyResults = (searchText?: string) => {
+  const onEmptyResults = useCallback((searchText?: string) => {
     setEmptyResultsWithSearch(!!searchText);
-  };
+  }, []);
 
   const onDbChange = ({ id: dbId }: { id: number }) => {
     setEmptyState?.(false);
@@ -152,7 +157,11 @@ const SqlEditorLeftBar = ({
     [tables],
   );
 
-  const onTablesChange = (tableNames: string[], schemaName: string) => {
+  const onTablesChange = (
+    tableNames: string[],
+    catalogName: string | null,
+    schemaName: string,
+  ) => {
     if (!schemaName) {
       return;
     }
@@ -169,7 +178,7 @@ const SqlEditorLeftBar = ({
     });
 
     tablesToAdd.forEach(tableName => {
-      dispatch(addTable(queryEditor, tableName, schemaName));
+      dispatch(addTable(queryEditor, tableName, catalogName, schemaName));
     });
 
     dispatch(removeTables(currentTables));
@@ -210,6 +219,15 @@ const SqlEditorLeftBar = ({
   const shouldShowReset = window.location.search === '?reset=1';
   const tableMetaDataHeight = height - 130; // 130 is the height of the selects above
 
+  const handleCatalogChange = useCallback(
+    (catalog: string | null) => {
+      if (queryEditor) {
+        dispatch(queryEditorSetCatalog(queryEditor, catalog));
+      }
+    },
+    [dispatch, queryEditor],
+  );
+
   const handleSchemaChange = useCallback(
     (schema: string) => {
       if (queryEditor) {
@@ -246,9 +264,11 @@ const SqlEditorLeftBar = ({
         getDbList={handleDbList}
         handleError={handleError}
         onDbChange={onDbChange}
+        onCatalogChange={handleCatalogChange}
+        catalog={catalog}
         onSchemaChange={handleSchemaChange}
-        onTableSelectChange={onTablesChange}
         schema={schema}
+        onTableSelectChange={onTablesChange}
         tableValue={selectedTableNames}
         sqlLabMode
       />
diff --git a/superset-frontend/src/SqlLab/components/TabbedSqlEditors/index.tsx b/superset-frontend/src/SqlLab/components/TabbedSqlEditors/index.tsx
index 078276ad26abf..7b4db1cbe844c 100644
--- a/superset-frontend/src/SqlLab/components/TabbedSqlEditors/index.tsx
+++ b/superset-frontend/src/SqlLab/components/TabbedSqlEditors/index.tsx
@@ -111,6 +111,7 @@ class TabbedSqlEditors extends React.PureComponent<TabbedSqlEditorsProps> {
       queryId,
       dbid,
       dbname,
+      catalog,
       schema,
       autorun,
       new: isNewQuery,
@@ -149,6 +150,7 @@ class TabbedSqlEditors extends React.PureComponent<TabbedSqlEditorsProps> {
         const newQueryEditor = {
           name,
           dbId: databaseId,
+          catalog,
           schema,
           autorun,
           sql,
diff --git a/superset-frontend/src/SqlLab/components/TableElement/index.tsx b/superset-frontend/src/SqlLab/components/TableElement/index.tsx
index e29a654c227c7..2f93491e88817 100644
--- a/superset-frontend/src/SqlLab/components/TableElement/index.tsx
+++ b/superset-frontend/src/SqlLab/components/TableElement/index.tsx
@@ -101,7 +101,7 @@ const StyledCollapsePanel = styled(Collapse.Panel)`
 `;
 
 const TableElement = ({ table, ...props }: TableElementProps) => {
-  const { dbId, schema, name, expanded } = table;
+  const { dbId, catalog, schema, name, expanded } = table;
   const theme = useTheme();
   const dispatch = useDispatch();
   const {
@@ -112,6 +112,7 @@ const TableElement = ({ table, ...props }: TableElementProps) => {
   } = useTableMetadataQuery(
     {
       dbId,
+      catalog,
       schema,
       table: name,
     },
@@ -125,6 +126,7 @@ const TableElement = ({ table, ...props }: TableElementProps) => {
   } = useTableExtendedMetadataQuery(
     {
       dbId,
+      catalog,
       schema,
       table: name,
     },
diff --git a/superset-frontend/src/SqlLab/fixtures.ts b/superset-frontend/src/SqlLab/fixtures.ts
index 742145c58ca6d..54f1c278f8f2b 100644
--- a/superset-frontend/src/SqlLab/fixtures.ts
+++ b/superset-frontend/src/SqlLab/fixtures.ts
@@ -36,6 +36,7 @@ export const table = {
   dbId: 1,
   selectStar: 'SELECT * FROM ab_user',
   queryEditorId: 'dfsadfs',
+  catalog: null,
   schema: 'superset',
   name: 'ab_user',
   id: 'r11Vgt60',
@@ -191,6 +192,7 @@ export const defaultQueryEditor = {
   selectedText: undefined,
   sql: 'SELECT *\nFROM\nWHERE',
   name: 'Untitled Query 1',
+  catalog: null,
   schema: 'main',
   remoteId: null,
   hideLeftBar: false,
@@ -233,6 +235,7 @@ export const queries = [
     queryLimit: 100,
     endDttm: 1476910566798,
     limit_reached: false,
+    catalog: null,
     schema: 'test_schema',
     errorMessage: null,
     db: 'main',
@@ -294,6 +297,7 @@ export const queries = [
     rows: 42,
     endDttm: 1476910579693,
     limit_reached: false,
+    catalog: null,
     schema: null,
     errorMessage: null,
     db: 'main',
@@ -323,6 +327,7 @@ export const queryWithNoQueryLimit = {
   rows: 42,
   endDttm: 1476910566798,
   limit_reached: false,
+  catalog: null,
   schema: 'test_schema',
   errorMessage: null,
   db: 'main',
@@ -456,18 +461,21 @@ export const tables = {
   options: [
     {
       value: 'birth_names',
+      catalog: null,
       schema: 'main',
       label: 'birth_names',
       title: 'birth_names',
     },
     {
       value: 'energy_usage',
+      catalog: null,
       schema: 'main',
       label: 'energy_usage',
       title: 'energy_usage',
     },
     {
       value: 'wb_health_population',
+      catalog: null,
       schema: 'main',
       label: 'wb_health_population',
       title: 'wb_health_population',
@@ -483,6 +491,7 @@ export const stoppedQuery = {
   progress: 0,
   results: [],
   runAsync: false,
+  catalog: null,
   schema: 'main',
   sql: 'SELECT ...',
   sqlEditorId: 'rJaf5u9WZ',
@@ -501,6 +510,7 @@ export const failedQueryWithErrorMessage = {
   progress: 0,
   results: [],
   runAsync: false,
+  catalog: null,
   schema: 'main',
   sql: 'SELECT ...',
   sqlEditorId: 'rJaf5u9WZ',
@@ -526,6 +536,7 @@ export const failedQueryWithErrors = {
   progress: 0,
   results: [],
   runAsync: false,
+  catalog: null,
   schema: 'main',
   sql: 'SELECT ...',
   sqlEditorId: 'rJaf5u9WZ',
@@ -555,6 +566,7 @@ const baseQuery: QueryResponse = {
   started: 'started',
   queryLimit: 100,
   endDttm: 1476910566798,
+  catalog: null,
   schema: 'test_schema',
   errorMessage: null,
   db: { key: 'main' },
@@ -689,6 +701,7 @@ export const query = {
   dbId: 1,
   sql: 'SELECT * FROM something',
   description: 'test description',
+  catalog: null,
   schema: 'test schema',
   resultsKey: 'test',
 };
@@ -698,6 +711,7 @@ export const queryId = 'clientId2353';
 export const testQuery: ISaveableDatasource = {
   name: 'unimportant',
   dbId: 1,
+  catalog: null,
   schema: 'main',
   sql: 'SELECT *',
   columns: [
@@ -727,6 +741,7 @@ export const mockdatasets = [...new Array(3)].map((_, i) => ({
   database_name: `db ${i}`,
   explore_url: `/explore/?datasource_type=table&datasource_id=${i}`,
   id: i,
+  catalog: null,
   schema: `schema ${i}`,
   table_name: `coolest table ${i}`,
   owners: [{ username: 'admin', userId: 1 }],
diff --git a/superset-frontend/src/SqlLab/reducers/getInitialState.ts b/superset-frontend/src/SqlLab/reducers/getInitialState.ts
index bcdc1f40c33da..52a9770854a06 100644
--- a/superset-frontend/src/SqlLab/reducers/getInitialState.ts
+++ b/superset-frontend/src/SqlLab/reducers/getInitialState.ts
@@ -89,6 +89,7 @@ export default function getInitialState({
         autorun: Boolean(activeTab.autorun),
         templateParams: activeTab.template_params || undefined,
         dbId: activeTab.database_id,
+        catalog: activeTab.catalog,
         schema: activeTab.schema,
         queryLimit: activeTab.query_limit,
         hideLeftBar: activeTab.hide_left_bar,
@@ -121,6 +122,7 @@ export default function getInitialState({
         const table = {
           dbId: tableSchema.database_id,
           queryEditorId: tableSchema.tab_state_id.toString(),
+          catalog: tableSchema.catalog,
           schema: tableSchema.schema,
           name: tableSchema.table,
           expanded: tableSchema.expanded,
diff --git a/superset-frontend/src/SqlLab/reducers/sqlLab.js b/superset-frontend/src/SqlLab/reducers/sqlLab.js
index 9ffcb4dfcb856..ecc0f090a9aaf 100644
--- a/superset-frontend/src/SqlLab/reducers/sqlLab.js
+++ b/superset-frontend/src/SqlLab/reducers/sqlLab.js
@@ -109,6 +109,7 @@ export default function sqlLabReducer(state = {}, action) {
         remoteId: progenitor.remoteId,
         name: t('Copy of %s', progenitor.name),
         dbId: action.query.dbId ? action.query.dbId : null,
+        catalog: action.query.catalog ? action.query.catalog : null,
         schema: action.query.schema ? action.query.schema : null,
         autorun: true,
         sql: action.query.sql,
@@ -180,6 +181,7 @@ export default function sqlLabReducer(state = {}, action) {
         if (
           xt.dbId === at.dbId &&
           xt.queryEditorId === at.queryEditorId &&
+          xt.catalog === at.catalog &&
           xt.schema === at.schema &&
           xt.name === at.name
         ) {
@@ -503,6 +505,18 @@ export default function sqlLabReducer(state = {}, action) {
         ),
       };
     },
+    [actions.QUERY_EDITOR_SET_CATALOG]() {
+      return {
+        ...state,
+        ...alterUnsavedQueryEditorState(
+          state,
+          {
+            catalog: action.catalog,
+          },
+          action.queryEditor.id,
+        ),
+      };
+    },
     [actions.QUERY_EDITOR_SET_SCHEMA]() {
       return {
         ...state,
diff --git a/superset-frontend/src/SqlLab/types.ts b/superset-frontend/src/SqlLab/types.ts
index cac9ceb5d91c4..b1dea6f2e31de 100644
--- a/superset-frontend/src/SqlLab/types.ts
+++ b/superset-frontend/src/SqlLab/types.ts
@@ -50,6 +50,7 @@ export interface QueryEditor {
   dbId?: number;
   name: string;
   title?: string; // keep it optional for backward compatibility
+  catalog?: string | null;
   schema?: string;
   autorun: boolean;
   sql: string;
@@ -81,6 +82,7 @@ export type UnsavedQueryEditor = Partial<QueryEditor>;
 export interface Table {
   id: string;
   dbId: number;
+  catalog: string | null;
   schema: string;
   name: string;
   queryEditorId: QueryEditor['id'];
diff --git a/superset-frontend/src/SqlLab/utils/reduxStateToLocalStorageHelper.ts b/superset-frontend/src/SqlLab/utils/reduxStateToLocalStorageHelper.ts
index 8b7f41f9f7481..3061de1f69b07 100644
--- a/superset-frontend/src/SqlLab/utils/reduxStateToLocalStorageHelper.ts
+++ b/superset-frontend/src/SqlLab/utils/reduxStateToLocalStorageHelper.ts
@@ -109,22 +109,24 @@ export function rehydratePersistedState(
   state: SqlLabRootState,
 ) {
   // Rehydrate server side persisted table metadata
-  state.sqlLab.tables.forEach(({ name: table, schema, dbId, persistData }) => {
-    if (dbId && schema && table && persistData?.columns) {
-      dispatch(
-        tableApiUtil.upsertQueryData(
-          'tableMetadata',
-          { dbId, schema, table },
-          persistData,
-        ),
-      );
-      dispatch(
-        tableApiUtil.upsertQueryData(
-          'tableExtendedMetadata',
-          { dbId, schema, table },
-          {},
-        ),
-      );
-    }
-  });
+  state.sqlLab.tables.forEach(
+    ({ name: table, catalog, schema, dbId, persistData }) => {
+      if (dbId && schema && table && persistData?.columns) {
+        dispatch(
+          tableApiUtil.upsertQueryData(
+            'tableMetadata',
+            { dbId, catalog, schema, table },
+            persistData,
+          ),
+        );
+        dispatch(
+          tableApiUtil.upsertQueryData(
+            'tableExtendedMetadata',
+            { dbId, catalog, schema, table },
+            {},
+          ),
+        );
+      }
+    },
+  );
 }
diff --git a/superset-frontend/src/components/DatabaseSelector/DatabaseSelector.test.tsx b/superset-frontend/src/components/DatabaseSelector/DatabaseSelector.test.tsx
index c3ad51cf60799..32373301f6030 100644
--- a/superset-frontend/src/components/DatabaseSelector/DatabaseSelector.test.tsx
+++ b/superset-frontend/src/components/DatabaseSelector/DatabaseSelector.test.tsx
@@ -40,6 +40,7 @@ const createProps = (): DatabaseSelectorProps => ({
   formMode: false,
   isDatabaseSelectEnabled: true,
   readOnly: false,
+  catalog: null,
   schema: 'public',
   sqlLabMode: true,
   getDbList: jest.fn(),
@@ -158,16 +159,23 @@ const fakeSchemaApiResult = {
   result: ['information_schema', 'public'],
 };
 
+const fakeCatalogApiResult = {
+  count: 0,
+  result: [],
+};
+
 const fakeFunctionNamesApiResult = {
   function_names: [],
 };
 
 const databaseApiRoute = 'glob:*/api/v1/database/?*';
+const catalogApiRoute = 'glob:*/api/v1/database/*/catalogs/?*';
 const schemaApiRoute = 'glob:*/api/v1/database/*/schemas/?*';
 const tablesApiRoute = 'glob:*/api/v1/database/*/tables/*';
 
 function setupFetchMock() {
   fetchMock.get(databaseApiRoute, fakeDatabaseApiResult);
+  fetchMock.get(catalogApiRoute, fakeCatalogApiResult);
   fetchMock.get(schemaApiRoute, fakeSchemaApiResult);
   fetchMock.get(tablesApiRoute, fakeFunctionNamesApiResult);
 }
diff --git a/superset-frontend/src/components/DatabaseSelector/index.tsx b/superset-frontend/src/components/DatabaseSelector/index.tsx
index 0c0268db5c9e6..6eb1340d5bcc3 100644
--- a/superset-frontend/src/components/DatabaseSelector/index.tsx
+++ b/superset-frontend/src/components/DatabaseSelector/index.tsx
@@ -24,7 +24,12 @@ import Label from 'src/components/Label';
 import { FormLabel } from 'src/components/Form';
 import RefreshLabel from 'src/components/RefreshLabel';
 import { useToasts } from 'src/components/MessageToasts/withToasts';
-import { useSchemas, SchemaOption } from 'src/hooks/apiResources';
+import {
+  useCatalogs,
+  CatalogOption,
+  useSchemas,
+  SchemaOption,
+} from 'src/hooks/apiResources';
 
 const DatabaseSelectorWrapper = styled.div`
   ${({ theme }) => `
@@ -81,6 +86,7 @@ export type DatabaseObject = {
   id: number;
   database_name: string;
   backend?: string;
+  allow_multi_catalog?: boolean;
 };
 
 export interface DatabaseSelectorProps {
@@ -92,9 +98,11 @@ export interface DatabaseSelectorProps {
   isDatabaseSelectEnabled?: boolean;
   onDbChange?: (db: DatabaseObject) => void;
   onEmptyResults?: (searchText?: string) => void;
+  onCatalogChange?: (catalog?: string) => void;
+  catalog?: string | null;
   onSchemaChange?: (schema?: string) => void;
-  readOnly?: boolean;
   schema?: string;
+  readOnly?: boolean;
   sqlLabMode?: boolean;
 }
 
@@ -113,6 +121,7 @@ const SelectLabel = ({
   </LabelStyle>
 );
 
+const EMPTY_CATALOG_OPTIONS: CatalogOption[] = [];
 const EMPTY_SCHEMA_OPTIONS: SchemaOption[] = [];
 
 export default function DatabaseSelector({
@@ -124,12 +133,20 @@ export default function DatabaseSelector({
   isDatabaseSelectEnabled = true,
   onDbChange,
   onEmptyResults,
+  onCatalogChange,
+  catalog,
   onSchemaChange,
-  readOnly = false,
   schema,
+  readOnly = false,
   sqlLabMode = false,
 }: DatabaseSelectorProps) {
+  const showCatalogSelector = !!db?.allow_multi_catalog;
   const [currentDb, setCurrentDb] = useState<DatabaseValue | undefined>();
+  const [currentCatalog, setCurrentCatalog] = useState<
+    CatalogOption | undefined
+  >(catalog ? { label: catalog, value: catalog, title: catalog } : undefined);
+  const catalogRef = useRef(catalog);
+  catalogRef.current = catalog;
   const [currentSchema, setCurrentSchema] = useState<SchemaOption | undefined>(
     schema ? { label: schema, value: schema, title: schema } : undefined,
   );
@@ -185,6 +202,7 @@ export default function DatabaseSelector({
             id: row.id,
             database_name: row.database_name,
             backend: row.backend,
+            allow_multi_catalog: row.allow_multi_catalog,
           }));
 
           return {
@@ -193,7 +211,7 @@ export default function DatabaseSelector({
           };
         });
       },
-    [formMode, getDbList, sqlLabMode],
+    [formMode, getDbList, sqlLabMode, onEmptyResults],
   );
 
   useEffect(() => {
@@ -223,11 +241,12 @@ export default function DatabaseSelector({
   }
 
   const {
-    data,
+    data: schemaData,
     isFetching: loadingSchemas,
-    refetch,
+    refetch: refetchSchemas,
   } = useSchemas({
     dbId: currentDb?.value,
+    catalog: currentCatalog?.value,
     onSuccess: (schemas, isFetched) => {
       if (schemas.length === 1) {
         changeSchema(schemas[0]);
@@ -244,17 +263,55 @@ export default function DatabaseSelector({
     onError: () => handleError(t('There was an error loading the schemas')),
   });
 
-  const schemaOptions = data || EMPTY_SCHEMA_OPTIONS;
+  const schemaOptions = schemaData || EMPTY_SCHEMA_OPTIONS;
+
+  function changeCatalog(catalog: CatalogOption | undefined) {
+    setCurrentCatalog(catalog);
+    setCurrentSchema(undefined);
+    if (onCatalogChange && catalog?.value !== catalogRef.current) {
+      onCatalogChange(catalog?.value);
+    }
+  }
+
+  const {
+    data: catalogData,
+    isFetching: loadingCatalogs,
+    refetch: refetchCatalogs,
+  } = useCatalogs({
+    dbId: currentDb?.value,
+    onSuccess: (catalogs, isFetched) => {
+      if (catalogs.length === 1) {
+        changeCatalog(catalogs[0]);
+      } else if (
+        !catalogs.find(
+          catalogOption => catalogRef.current === catalogOption.value,
+        )
+      ) {
+        changeCatalog(undefined);
+      }
+
+      if (isFetched) {
+        addSuccessToast('List refreshed');
+      }
+    },
+    onError: () => handleError(t('There was an error loading the catalogs')),
+  });
+
+  const catalogOptions = catalogData || EMPTY_CATALOG_OPTIONS;
 
-  function changeDataBase(
+  function changeDatabase(
     value: { label: string; value: number },
     database: DatabaseValue,
   ) {
     setCurrentDb(database);
+    setCurrentCatalog(undefined);
     setCurrentSchema(undefined);
     if (onDbChange) {
       onDbChange(database);
     }
+    if (onCatalogChange) {
+      onCatalogChange(undefined);
+    }
     if (onSchemaChange) {
       onSchemaChange(undefined);
     }
@@ -278,7 +335,7 @@ export default function DatabaseSelector({
         header={<FormLabel>{t('Database')}</FormLabel>}
         lazyLoading={false}
         notFoundContent={emptyState}
-        onChange={changeDataBase}
+        onChange={changeDatabase}
         value={currentDb}
         placeholder={t('Select database or type to search databases')}
         disabled={!isDatabaseSelectEnabled || readOnly}
@@ -288,10 +345,36 @@ export default function DatabaseSelector({
     );
   }
 
+  function renderCatalogSelect() {
+    const refreshIcon = !readOnly && (
+      <RefreshLabel
+        onClick={refetchCatalogs}
+        tooltipContent={t('Force refresh catalog list')}
+      />
+    );
+    return renderSelectRow(
+      <Select
+        ariaLabel={t('Select catalog or type to search catalogs')}
+        disabled={!currentDb || readOnly}
+        header={<FormLabel>{t('Catalog')}</FormLabel>}
+        labelInValue
+        loading={loadingCatalogs}
+        name="select-catalog"
+        notFoundContent={t('No compatible catalog found')}
+        placeholder={t('Select catalog or type to search catalogs')}
+        onChange={item => changeCatalog(item as CatalogOption)}
+        options={catalogOptions}
+        showSearch
+        value={currentCatalog}
+      />,
+      refreshIcon,
+    );
+  }
+
   function renderSchemaSelect() {
     const refreshIcon = !readOnly && (
       <RefreshLabel
-        onClick={() => refetch()}
+        onClick={refetchSchemas}
         tooltipContent={t('Force refresh schema list')}
       />
     );
@@ -317,6 +400,7 @@ export default function DatabaseSelector({
   return (
     <DatabaseSelectorWrapper data-test="DatabaseSelector">
       {renderDatabaseSelect()}
+      {showCatalogSelector && renderCatalogSelect()}
       {renderSchemaSelect()}
     </DatabaseSelectorWrapper>
   );
diff --git a/superset-frontend/src/components/Datasource/DatasourceEditor.jsx b/superset-frontend/src/components/Datasource/DatasourceEditor.jsx
index b9af27d2af186..1aa20312f76b4 100644
--- a/superset-frontend/src/components/Datasource/DatasourceEditor.jsx
+++ b/superset-frontend/src/components/Datasource/DatasourceEditor.jsx
@@ -758,6 +758,7 @@ class DatasourceEditor extends React.PureComponent {
       datasource_type: datasource.type || datasource.datasource_type,
       database_name:
         datasource.database.database_name || datasource.database.name,
+      catalog_name: datasource.catalog,
       schema_name: datasource.schema,
       table_name: datasource.table_name,
       normalize_columns: datasource.normalize_columns,
@@ -1090,7 +1091,12 @@ class DatasourceEditor extends React.PureComponent {
                         <div css={{ marginTop: 8 }}>
                           <DatabaseSelector
                             db={datasource?.database}
+                            catalog={datasource.catalog}
                             schema={datasource.schema}
+                            onCatalogChange={catalog =>
+                              this.state.isEditMode &&
+                              this.onDatasourcePropChange('catalog', catalog)
+                            }
                             onSchemaChange={schema =>
                               this.state.isEditMode &&
                               this.onDatasourcePropChange('schema', schema)
@@ -1164,9 +1170,16 @@ class DatasourceEditor extends React.PureComponent {
                         }}
                         dbId={datasource.database?.id}
                         handleError={this.props.addDangerToast}
+                        catalog={datasource.catalog}
                         schema={datasource.schema}
                         sqlLabMode={false}
                         tableValue={datasource.table_name}
+                        onCatalogChange={
+                          this.state.isEditMode
+                            ? catalog =>
+                                this.onDatasourcePropChange('catalog', catalog)
+                            : undefined
+                        }
                         onSchemaChange={
                           this.state.isEditMode
                             ? schema =>
diff --git a/superset-frontend/src/components/TableSelector/TableSelector.test.tsx b/superset-frontend/src/components/TableSelector/TableSelector.test.tsx
index 13a2c33248bea..8f82818b8e16c 100644
--- a/superset-frontend/src/components/TableSelector/TableSelector.test.tsx
+++ b/superset-frontend/src/components/TableSelector/TableSelector.test.tsx
@@ -54,6 +54,7 @@ const getTableMockFunction = () =>
   }) as any;
 
 const databaseApiRoute = 'glob:*/api/v1/database/?*';
+const catalogApiRoute = 'glob:*/api/v1/database/*/catalogs/?*';
 const schemaApiRoute = 'glob:*/api/v1/database/*/schemas/?*';
 const tablesApiRoute = 'glob:*/api/v1/database/*/tables/*';
 
@@ -74,6 +75,7 @@ afterEach(() => {
 });
 
 test('renders with default props', async () => {
+  fetchMock.get(catalogApiRoute, { result: [] });
   fetchMock.get(schemaApiRoute, { result: [] });
   fetchMock.get(tablesApiRoute, getTableMockFunction());
 
@@ -96,6 +98,7 @@ test('renders with default props', async () => {
 });
 
 test('skips select all options', async () => {
+  fetchMock.get(catalogApiRoute, { result: [] });
   fetchMock.get(schemaApiRoute, { result: ['test_schema'] });
   fetchMock.get(tablesApiRoute, getTableMockFunction());
 
@@ -115,6 +118,7 @@ test('skips select all options', async () => {
 });
 
 test('renders table options without Select All option', async () => {
+  fetchMock.get(catalogApiRoute, { result: [] });
   fetchMock.get(schemaApiRoute, { result: ['test_schema'] });
   fetchMock.get(tablesApiRoute, getTableMockFunction());
 
@@ -133,6 +137,7 @@ test('renders table options without Select All option', async () => {
 });
 
 test('renders disabled without schema', async () => {
+  fetchMock.get(catalogApiRoute, { result: [] });
   fetchMock.get(schemaApiRoute, { result: [] });
   fetchMock.get(tablesApiRoute, getTableMockFunction());
 
@@ -150,6 +155,7 @@ test('renders disabled without schema', async () => {
 });
 
 test('table select retain value if not in SQL Lab mode', async () => {
+  fetchMock.get(catalogApiRoute, { result: [] });
   fetchMock.get(schemaApiRoute, { result: ['test_schema'] });
   fetchMock.get(tablesApiRoute, getTableMockFunction());
 
@@ -191,6 +197,7 @@ test('table select retain value if not in SQL Lab mode', async () => {
 });
 
 test('table multi select retain all the values selected', async () => {
+  fetchMock.get(catalogApiRoute, { result: [] });
   fetchMock.get(schemaApiRoute, { result: ['test_schema'] });
   fetchMock.get(tablesApiRoute, getTableMockFunction());
 
diff --git a/superset-frontend/src/components/TableSelector/index.tsx b/superset-frontend/src/components/TableSelector/index.tsx
index b7c92a978a6e9..05e3808c495cd 100644
--- a/superset-frontend/src/components/TableSelector/index.tsx
+++ b/superset-frontend/src/components/TableSelector/index.tsx
@@ -97,13 +97,19 @@ interface TableSelectorProps {
   handleError: (msg: string) => void;
   isDatabaseSelectEnabled?: boolean;
   onDbChange?: (db: DatabaseObject) => void;
+  onCatalogChange?: (catalog?: string | null) => void;
   onSchemaChange?: (schema?: string) => void;
   readOnly?: boolean;
+  catalog?: string | null;
   schema?: string;
   onEmptyResults?: (searchText?: string) => void;
   sqlLabMode?: boolean;
   tableValue?: string | string[];
-  onTableSelectChange?: (value?: string | string[], schema?: string) => void;
+  onTableSelectChange?: (
+    value?: string | string[],
+    catalog?: string | null,
+    schema?: string,
+  ) => void;
   tableSelectMode?: 'single' | 'multiple';
   customTableOptionLabelRenderer?: (table: Table) => JSX.Element;
 }
@@ -159,9 +165,11 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
   handleError,
   isDatabaseSelectEnabled = true,
   onDbChange,
+  onCatalogChange,
   onSchemaChange,
   readOnly = false,
   onEmptyResults,
+  catalog,
   schema,
   sqlLabMode = true,
   tableSelectMode = 'single',
@@ -170,6 +178,9 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
   customTableOptionLabelRenderer,
 }) => {
   const { addSuccessToast } = useToasts();
+  const [currentCatalog, setCurrentCatalog] = useState<
+    string | null | undefined
+  >(catalog);
   const [currentSchema, setCurrentSchema] = useState<string | undefined>(
     schema,
   );
@@ -182,6 +193,7 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
     refetch,
   } = useTables({
     dbId: database?.id,
+    catalog: currentCatalog,
     schema: currentSchema,
     onSuccess: (data, isFetched) => {
       if (isFetched) {
@@ -218,6 +230,7 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
   useEffect(() => {
     // reset selections
     if (database === undefined) {
+      setCurrentCatalog(undefined);
       setCurrentSchema(undefined);
       setTableSelectValue(undefined);
     }
@@ -245,6 +258,7 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
         Array.isArray(selectedOptions)
           ? selectedOptions.map(option => option?.value)
           : selectedOptions?.value,
+        currentCatalog,
         currentSchema,
       );
     } else {
@@ -256,6 +270,22 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
     if (onDbChange) {
       onDbChange(db);
     }
+
+    setCurrentCatalog(undefined);
+    setCurrentSchema(undefined);
+    const value = tableSelectMode === 'single' ? undefined : [];
+    setTableSelectValue(value);
+  };
+
+  const internalCatalogChange = (catalog?: string | null) => {
+    setCurrentCatalog(catalog);
+    if (onCatalogChange) {
+      onCatalogChange(catalog);
+    }
+
+    setCurrentSchema(undefined);
+    const value = tableSelectMode === 'single' ? undefined : [];
+    setTableSelectValue(value);
   };
 
   const internalSchemaChange = (schema?: string) => {
@@ -265,7 +295,7 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
     }
 
     const value = tableSelectMode === 'single' ? undefined : [];
-    internalTableChange(value);
+    setTableSelectValue(value);
   };
 
   const handleFilterOption = useMemo(
@@ -328,6 +358,8 @@ const TableSelector: FunctionComponent<TableSelectorProps> = ({
         handleError={handleError}
         onDbChange={readOnly ? undefined : internalDbChange}
         onEmptyResults={onEmptyResults}
+        onCatalogChange={readOnly ? undefined : internalCatalogChange}
+        catalog={currentCatalog}
         onSchemaChange={readOnly ? undefined : internalSchemaChange}
         schema={currentSchema}
         sqlLabMode={sqlLabMode}
diff --git a/superset-frontend/src/explore/components/controls/DndColumnSelectControl/DndFilterSelect.tsx b/superset-frontend/src/explore/components/controls/DndColumnSelectControl/DndFilterSelect.tsx
index 696861b28913d..55cd1c607181a 100644
--- a/superset-frontend/src/explore/components/controls/DndColumnSelectControl/DndFilterSelect.tsx
+++ b/superset-frontend/src/explore/components/controls/DndColumnSelectControl/DndFilterSelect.tsx
@@ -48,6 +48,7 @@ import {
 } from 'src/explore/components/DatasourcePanel/types';
 import { DndItemType } from 'src/explore/components/DndItemType';
 import { ControlComponentProps } from 'src/explore/components/Control';
+import { toQueryString } from 'src/utils/urlUtils';
 import DndAdhocFilterOption from './DndAdhocFilterOption';
 import { useDefaultTimeFilter } from '../DateFilterControl/utils';
 import { Clauses, ExpressionTypes } from '../FilterControl/types';
@@ -175,13 +176,20 @@ const DndFilterSelect = (props: DndFilterSelectProps) => {
       const dbId = datasource.database?.id;
       const {
         datasource_name: name,
+        catalog,
         schema,
         is_sqllab_view: isSqllabView,
       } = datasource;
 
       if (!isSqllabView && dbId && name && schema) {
         SupersetClient.get({
-          endpoint: `/api/v1/database/${dbId}/table_metadata/extra/?name=${name}&schema=${schema}`,
+          endpoint: `/api/v1/database/${dbId}/table_metadata/extra/${toQueryString(
+            {
+              name,
+              catalog,
+              schema,
+            },
+          )}`,
         })
           .then(({ json }: { json: Record<string, any> }) => {
             if (json?.partitions) {
diff --git a/superset-frontend/src/explore/components/controls/FilterControl/AdhocFilterControl/index.jsx b/superset-frontend/src/explore/components/controls/FilterControl/AdhocFilterControl/index.jsx
index c68a97b4d728f..f4f604ecf0526 100644
--- a/superset-frontend/src/explore/components/controls/FilterControl/AdhocFilterControl/index.jsx
+++ b/superset-frontend/src/explore/components/controls/FilterControl/AdhocFilterControl/index.jsx
@@ -48,6 +48,7 @@ import AdhocFilterOption from 'src/explore/components/controls/FilterControl/Adh
 import AdhocFilter from 'src/explore/components/controls/FilterControl/AdhocFilter';
 import adhocFilterType from 'src/explore/components/controls/FilterControl/adhocFilterType';
 import columnType from 'src/explore/components/controls/FilterControl/columnType';
+import { toQueryString } from 'src/utils/urlUtils';
 import { Clauses, ExpressionTypes } from '../types';
 
 const { warning } = Modal;
@@ -137,13 +138,20 @@ class AdhocFilterControl extends React.Component {
       const dbId = datasource.database?.id;
       const {
         datasource_name: name,
+        catalog,
         schema,
         is_sqllab_view: isSqllabView,
       } = datasource;
 
       if (!isSqllabView && dbId && name && schema) {
         SupersetClient.get({
-          endpoint: `/api/v1/database/${dbId}/table_metadata/extra/?name=${name}&schema=${schema}`,
+          endpoint: `/api/v1/database/${dbId}/table_metadata/extra/${toQueryString(
+            {
+              name,
+              catalog,
+              schema,
+            },
+          )}`,
         })
           .then(({ json }) => {
             if (json && json.partitions) {
diff --git a/superset-frontend/src/explore/types.ts b/superset-frontend/src/explore/types.ts
index 51b3013233ea6..8ff0219fe6432 100644
--- a/superset-frontend/src/explore/types.ts
+++ b/superset-frontend/src/explore/types.ts
@@ -66,6 +66,7 @@ export type OptionSortType = Partial<
 export type Datasource = Dataset & {
   database?: DatabaseObject;
   datasource?: string;
+  catalog?: string | null;
   schema?: string;
   is_sqllab_view?: boolean;
   extra?: string;
diff --git a/superset-frontend/src/features/databases/DatabaseModal/ExtraOptions.tsx b/superset-frontend/src/features/databases/DatabaseModal/ExtraOptions.tsx
index 28e02a02f0c43..5700b59c85183 100644
--- a/superset-frontend/src/features/databases/DatabaseModal/ExtraOptions.tsx
+++ b/superset-frontend/src/features/databases/DatabaseModal/ExtraOptions.tsx
@@ -56,6 +56,8 @@ const ExtraOptions = ({
   const createAsOpen = !!(db?.allow_ctas || db?.allow_cvas);
   const isFileUploadSupportedByEngine =
     db?.engine_information?.supports_file_upload;
+  const supportsDynamicCatalog =
+    db?.engine_information?.supports_dynamic_catalog;
 
   // JSON.parse will deep parse engine_params
   // if it's an object, and we want to keep it a string
@@ -191,7 +193,8 @@ const ExtraOptions = ({
                 <IndeterminateCheckbox
                   id="allows_virtual_table_explore"
                   indeterminate={false}
-                  checked={!!extraJson?.allows_virtual_table_explore}
+                  // when `allows_virtual_table_explore` is not present in `extra` it defaults to true
+                  checked={extraJson?.allows_virtual_table_explore !== false}
                   onChange={onExtraInputChange}
                   labelText={t('Allow this database to be explored')}
                 />
@@ -587,6 +590,24 @@ const ExtraOptions = ({
             />
           </div>
         </StyledInputContainer>
+        {supportsDynamicCatalog && (
+          <StyledInputContainer css={no_margin_bottom}>
+            <div className="input-container">
+              <IndeterminateCheckbox
+                id="allow_multi_catalog"
+                indeterminate={false}
+                checked={!!extraJson?.allow_multi_catalog}
+                onChange={onExtraInputChange}
+                labelText={t('Allow changing catalogs')}
+              />
+              <InfoTooltip
+                tooltip={t(
+                  'Give access to multiple catalogs in a single database connection.',
+                )}
+              />
+            </div>
+          </StyledInputContainer>
+        )}
       </Collapse.Panel>
     </Collapse>
   );
diff --git a/superset-frontend/src/features/databases/DatabaseModal/SSHTunnelSwitch.test.tsx b/superset-frontend/src/features/databases/DatabaseModal/SSHTunnelSwitch.test.tsx
index fef205acf2ead..0ca40db0e9db4 100644
--- a/superset-frontend/src/features/databases/DatabaseModal/SSHTunnelSwitch.test.tsx
+++ b/superset-frontend/src/features/databases/DatabaseModal/SSHTunnelSwitch.test.tsx
@@ -96,6 +96,7 @@ test('Does not render if SSH Tunnel is disabled', () => {
         engine_information: {
           disable_ssh_tunneling: true,
           supports_file_upload: false,
+          supports_dynamic_catalog: false,
         },
       }}
     />,
diff --git a/superset-frontend/src/features/databases/UploadDataModel/UploadDataModal.test.tsx b/superset-frontend/src/features/databases/UploadDataModel/UploadDataModal.test.tsx
index 44a8a60738a70..ce5efd09cbf22 100644
--- a/superset-frontend/src/features/databases/UploadDataModel/UploadDataModal.test.tsx
+++ b/superset-frontend/src/features/databases/UploadDataModel/UploadDataModal.test.tsx
@@ -47,6 +47,10 @@ fetchMock.get(
   },
 );
 
+fetchMock.get('glob:*api/v1/database/*/catalogs/', {
+  result: [],
+});
+
 fetchMock.get('glob:*api/v1/database/1/schemas/', {
   result: ['information_schema', 'public'],
 });
diff --git a/superset-frontend/src/features/databases/types.ts b/superset-frontend/src/features/databases/types.ts
index 06799ebaface1..a09ad174a0474 100644
--- a/superset-frontend/src/features/databases/types.ts
+++ b/superset-frontend/src/features/databases/types.ts
@@ -109,6 +109,7 @@ export type DatabaseObject = {
   engine_information?: {
     supports_file_upload?: boolean;
     disable_ssh_tunneling?: boolean;
+    supports_dynamic_catalog?: boolean;
   };
 
   // SSH Tunnel information
@@ -202,6 +203,7 @@ export type DatabaseForm = {
   engine_information: {
     supports_file_upload: boolean;
     disable_ssh_tunneling: boolean;
+    supports_dynamic_catalog: boolean;
   };
 };
 
@@ -223,6 +225,7 @@ export interface ExtraJson {
   cost_estimate_enabled?: boolean; // in SQL Lab
   disable_data_preview?: boolean; // in SQL Lab
   disable_drill_to_detail?: boolean;
+  allow_multi_catalog?: boolean;
   engine_params?: {
     catalog?: Record<string, string>;
     connect_args?: {
diff --git a/superset-frontend/src/features/datasets/AddDataset/DatasetPanel/index.tsx b/superset-frontend/src/features/datasets/AddDataset/DatasetPanel/index.tsx
index b3f8aec8f99a0..d4e5c98104224 100644
--- a/superset-frontend/src/features/datasets/AddDataset/DatasetPanel/index.tsx
+++ b/superset-frontend/src/features/datasets/AddDataset/DatasetPanel/index.tsx
@@ -20,6 +20,7 @@ import React, { useEffect, useState, useRef } from 'react';
 import { SupersetClient, logging, t } from '@superset-ui/core';
 import { DatasetObject } from 'src/features/datasets/AddDataset/types';
 import { addDangerToast } from 'src/components/MessageToasts/actions';
+import { toQueryString } from 'src/utils/urlUtils';
 import DatasetPanel from './DatasetPanel';
 import { ITableColumn, IDatabaseTable, isIDatabaseTable } from './types';
 
@@ -51,8 +52,9 @@ export interface IDatasetPanelWrapperProps {
    */
   dbId?: number;
   /**
-   * The selected schema for the database
+   * The selected catalog/schema for the database
    */
+  catalog?: string | null;
   schema?: string | null;
   setHasColumns?: Function;
   datasets?: DatasetObject[] | undefined;
@@ -61,6 +63,7 @@ export interface IDatasetPanelWrapperProps {
 const DatasetPanelWrapper = ({
   tableName,
   dbId,
+  catalog,
   schema,
   setHasColumns,
   datasets,
@@ -74,9 +77,11 @@ const DatasetPanelWrapper = ({
     const { dbId, tableName, schema } = props;
     setLoading(true);
     setHasColumns?.(false);
-    const path = schema
-      ? `/api/v1/database/${dbId}/table_metadata/?name=${tableName}&schema=${schema}`
-      : `/api/v1/database/${dbId}/table_metadata/?name=${tableName}`;
+    const path = `/api/v1/database/${dbId}/table_metadata/${toQueryString({
+      name: tableName,
+      catalog,
+      schema,
+    })}`;
     try {
       const response = await SupersetClient.get({
         endpoint: path,
diff --git a/superset-frontend/src/features/datasets/AddDataset/Footer/index.tsx b/superset-frontend/src/features/datasets/AddDataset/Footer/index.tsx
index e0853cdc9d587..c4a5df31c4e8f 100644
--- a/superset-frontend/src/features/datasets/AddDataset/Footer/index.tsx
+++ b/superset-frontend/src/features/datasets/AddDataset/Footer/index.tsx
@@ -90,6 +90,7 @@ function Footer({
     if (datasetObject) {
       const data = {
         database: datasetObject.db?.id,
+        catalog: datasetObject.catalog,
         schema: datasetObject.schema,
         table_name: datasetObject.table_name,
       };
diff --git a/superset-frontend/src/features/datasets/AddDataset/LeftPanel/index.tsx b/superset-frontend/src/features/datasets/AddDataset/LeftPanel/index.tsx
index 093a65626ba49..d722f82b0f75c 100644
--- a/superset-frontend/src/features/datasets/AddDataset/LeftPanel/index.tsx
+++ b/superset-frontend/src/features/datasets/AddDataset/LeftPanel/index.tsx
@@ -129,6 +129,14 @@ export default function LeftPanel({
     },
     [setDataset],
   );
+  const setCatalog = (catalog: string | null) => {
+    if (catalog) {
+      setDataset({
+        type: DatasetActionType.SelectCatalog,
+        payload: { name: 'catalog', value: catalog },
+      });
+    }
+  };
   const setSchema = (schema: string) => {
     if (schema) {
       setDataset({
@@ -178,10 +186,12 @@ export default function LeftPanel({
         handleError={addDangerToast}
         emptyState={emptyStateComponent(false)}
         onDbChange={setDatabase}
+        onCatalogChange={setCatalog}
         onSchemaChange={setSchema}
         onTableSelectChange={setTable}
         sqlLabMode={false}
         customTableOptionLabelRenderer={customTableOptionLabelRenderer}
+        {...(dataset?.catalog && { catalog: dataset.catalog })}
         {...(dataset?.schema && { schema: dataset.schema })}
       />
     </LeftPanelStyle>
diff --git a/superset-frontend/src/features/datasets/AddDataset/types.tsx b/superset-frontend/src/features/datasets/AddDataset/types.tsx
index 998d315940ed0..1c4b16436d3cd 100644
--- a/superset-frontend/src/features/datasets/AddDataset/types.tsx
+++ b/superset-frontend/src/features/datasets/AddDataset/types.tsx
@@ -20,6 +20,7 @@ import { DatabaseObject } from 'src/components/DatabaseSelector';
 
 export enum DatasetActionType {
   SelectDatabase,
+  SelectCatalog,
   SelectSchema,
   SelectTable,
   ChangeDataset,
@@ -27,6 +28,7 @@ export enum DatasetActionType {
 
 export interface DatasetObject {
   db: DatabaseObject & { owners: [number] };
+  catalog?: string | null;
   schema?: string | null;
   dataset_name: string;
   table_name?: string | null;
@@ -50,6 +52,7 @@ export type DSReducerActionType =
   | {
       type:
         | DatasetActionType.ChangeDataset
+        | DatasetActionType.SelectCatalog
         | DatasetActionType.SelectSchema
         | DatasetActionType.SelectTable;
       payload: DatasetReducerPayloadType;
diff --git a/superset-frontend/src/hooks/apiResources/catalogs.ts b/superset-frontend/src/hooks/apiResources/catalogs.ts
new file mode 100644
index 0000000000000..1e6a97b34463b
--- /dev/null
+++ b/superset-frontend/src/hooks/apiResources/catalogs.ts
@@ -0,0 +1,127 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useCallback, useEffect, useRef } from 'react';
+import useEffectEvent from 'src/hooks/useEffectEvent';
+import { api, JsonResponse } from './queryApi';
+
+export type CatalogOption = {
+  value: string;
+  label: string;
+  title: string;
+};
+
+export type FetchCatalogsQueryParams = {
+  dbId?: string | number;
+  forceRefresh: boolean;
+  onSuccess?: (data: CatalogOption[], isRefetched: boolean) => void;
+  onError?: () => void;
+};
+
+type Params = Omit<FetchCatalogsQueryParams, 'forceRefresh'>;
+
+const catalogApi = api.injectEndpoints({
+  endpoints: builder => ({
+    catalogs: builder.query<CatalogOption[], FetchCatalogsQueryParams>({
+      providesTags: [{ type: 'Catalogs', id: 'LIST' }],
+      query: ({ dbId, forceRefresh }) => ({
+        endpoint: `/api/v1/database/${dbId}/catalogs/`,
+        urlParams: {
+          force: forceRefresh,
+        },
+        transformResponse: ({ json }: JsonResponse) =>
+          json.result.sort().map((value: string) => ({
+            value,
+            label: value,
+            title: value,
+          })),
+      }),
+      serializeQueryArgs: ({ queryArgs: { dbId } }) => ({
+        dbId,
+      }),
+    }),
+  }),
+});
+
+export const {
+  useLazyCatalogsQuery,
+  useCatalogsQuery,
+  endpoints: catalogEndpoints,
+  util: catalogApiUtil,
+} = catalogApi;
+
+export const EMPTY_CATALOGS = [] as CatalogOption[];
+
+export function useCatalogs(options: Params) {
+  const isMountedRef = useRef(false);
+  const { dbId, onSuccess, onError } = options || {};
+  const [trigger] = useLazyCatalogsQuery();
+  const result = useCatalogsQuery(
+    { dbId, forceRefresh: false },
+    {
+      skip: !dbId,
+    },
+  );
+
+  const handleOnSuccess = useEffectEvent(
+    (data: CatalogOption[], isRefetched: boolean) => {
+      onSuccess?.(data, isRefetched);
+    },
+  );
+
+  const handleOnError = useEffectEvent(() => {
+    onError?.();
+  });
+
+  const refetch = useCallback(() => {
+    if (dbId) {
+      trigger({ dbId, forceRefresh: true }).then(
+        ({ isSuccess, isError, data }) => {
+          if (isSuccess) {
+            handleOnSuccess(data || EMPTY_CATALOGS, true);
+          }
+          if (isError) {
+            handleOnError();
+          }
+        },
+      );
+    }
+  }, [dbId, handleOnError, handleOnSuccess, trigger]);
+
+  useEffect(() => {
+    if (isMountedRef.current) {
+      const { requestId, isSuccess, isError, isFetching, data, originalArgs } =
+        result;
+      if (!originalArgs?.forceRefresh && requestId && !isFetching) {
+        if (isSuccess) {
+          handleOnSuccess(data || EMPTY_CATALOGS, false);
+        }
+        if (isError) {
+          handleOnError();
+        }
+      }
+    } else {
+      isMountedRef.current = true;
+    }
+  }, [result, handleOnSuccess, handleOnError]);
+
+  return {
+    ...result,
+    refetch,
+  };
+}
diff --git a/superset-frontend/src/hooks/apiResources/index.ts b/superset-frontend/src/hooks/apiResources/index.ts
index faf4e5736b915..53aa7aa113cb5 100644
--- a/superset-frontend/src/hooks/apiResources/index.ts
+++ b/superset-frontend/src/hooks/apiResources/index.ts
@@ -26,6 +26,7 @@ export {
 // A central catalog of API Resource hooks.
 // Add new API hooks here, organized under
 // different files for different resource types.
+export * from './catalogs';
 export * from './charts';
 export * from './dashboards';
 export * from './tables';
diff --git a/superset-frontend/src/hooks/apiResources/queryApi.ts b/superset-frontend/src/hooks/apiResources/queryApi.ts
index 439418529fd98..b0c0bf5781b37 100644
--- a/superset-frontend/src/hooks/apiResources/queryApi.ts
+++ b/superset-frontend/src/hooks/apiResources/queryApi.ts
@@ -72,6 +72,7 @@ export const supersetClientQuery: BaseQueryFn<
 export const api = createApi({
   reducerPath: 'queryApi',
   tagTypes: [
+    'Catalogs',
     'Schemas',
     'Tables',
     'DatabaseFunctions',
diff --git a/superset-frontend/src/hooks/apiResources/queryValidations.ts b/superset-frontend/src/hooks/apiResources/queryValidations.ts
index 722c320049013..df88d6afca9bb 100644
--- a/superset-frontend/src/hooks/apiResources/queryValidations.ts
+++ b/superset-frontend/src/hooks/apiResources/queryValidations.ts
@@ -20,6 +20,7 @@ import { api, JsonResponse } from './queryApi';
 
 export type FetchValidationQueryParams = {
   dbId?: string | number;
+  catalog?: string | null;
   schema?: string;
   sql: string;
   templateParams?: string;
@@ -39,7 +40,7 @@ const queryValidationApi = api.injectEndpoints({
       FetchValidationQueryParams
     >({
       providesTags: ['QueryValidations'],
-      query: ({ dbId, schema, sql, templateParams }) => {
+      query: ({ dbId, catalog, schema, sql, templateParams }) => {
         let template_params = templateParams;
         try {
           template_params = JSON.parse(templateParams || '');
@@ -47,6 +48,7 @@ const queryValidationApi = api.injectEndpoints({
           template_params = undefined;
         }
         const postPayload = {
+          catalog,
           schema,
           sql,
           ...(template_params && { template_params }),
diff --git a/superset-frontend/src/hooks/apiResources/schemas.test.ts b/superset-frontend/src/hooks/apiResources/schemas.test.ts
index 70c1289d576a1..62e154acbe09d 100644
--- a/superset-frontend/src/hooks/apiResources/schemas.test.ts
+++ b/superset-frontend/src/hooks/apiResources/schemas.test.ts
@@ -80,7 +80,7 @@ describe('useSchemas hook', () => {
         })}`,
       ).length,
     ).toBe(1);
-    expect(onSuccess).toHaveBeenCalledTimes(1);
+    expect(onSuccess).toHaveBeenCalledTimes(2);
     act(() => {
       result.current.refetch();
     });
@@ -92,7 +92,7 @@ describe('useSchemas hook', () => {
         })}`,
       ).length,
     ).toBe(1);
-    expect(onSuccess).toHaveBeenCalledTimes(2);
+    expect(onSuccess).toHaveBeenCalledTimes(3);
     expect(result.current.data).toEqual(expectedResult);
   });
 
@@ -143,17 +143,17 @@ describe('useSchemas hook', () => {
 
     await waitFor(() => expect(result.current.data).toEqual(expectedResult));
     expect(fetchMock.calls(schemaApiRoute).length).toBe(1);
-    expect(onSuccess).toHaveBeenCalledTimes(1);
+    expect(onSuccess).toHaveBeenCalledTimes(2);
 
     rerender({ dbId: 'db2' });
     await waitFor(() => expect(result.current.data).toEqual(expectedResult2));
     expect(fetchMock.calls(schemaApiRoute).length).toBe(2);
-    expect(onSuccess).toHaveBeenCalledTimes(2);
+    expect(onSuccess).toHaveBeenCalledTimes(4);
 
     rerender({ dbId: expectDbId });
     await waitFor(() => expect(result.current.data).toEqual(expectedResult));
     expect(fetchMock.calls(schemaApiRoute).length).toBe(2);
-    expect(onSuccess).toHaveBeenCalledTimes(3);
+    expect(onSuccess).toHaveBeenCalledTimes(5);
 
     // clean up cache
     act(() => {
diff --git a/superset-frontend/src/hooks/apiResources/schemas.ts b/superset-frontend/src/hooks/apiResources/schemas.ts
index bbce48d5ad742..d5cec1daea298 100644
--- a/superset-frontend/src/hooks/apiResources/schemas.ts
+++ b/superset-frontend/src/hooks/apiResources/schemas.ts
@@ -28,6 +28,7 @@ export type SchemaOption = {
 
 export type FetchSchemasQueryParams = {
   dbId?: string | number;
+  catalog?: string;
   forceRefresh: boolean;
   onSuccess?: (data: SchemaOption[], isRefetched: boolean) => void;
   onError?: () => void;
@@ -39,14 +40,15 @@ const schemaApi = api.injectEndpoints({
   endpoints: builder => ({
     schemas: builder.query<SchemaOption[], FetchSchemasQueryParams>({
       providesTags: [{ type: 'Schemas', id: 'LIST' }],
-      query: ({ dbId, forceRefresh }) => ({
+      query: ({ dbId, catalog, forceRefresh }) => ({
         endpoint: `/api/v1/database/${dbId}/schemas/`,
         // TODO: Would be nice to add pagination in a follow-up. Needs endpoint changes.
         urlParams: {
           force: forceRefresh,
+          ...(catalog !== undefined && { catalog }),
         },
         transformResponse: ({ json }: JsonResponse) =>
-          json.result.map((value: string) => ({
+          json.result.sort().map((value: string) => ({
             value,
             label: value,
             title: value,
@@ -70,10 +72,10 @@ export const EMPTY_SCHEMAS = [] as SchemaOption[];
 
 export function useSchemas(options: Params) {
   const isMountedRef = useRef(false);
-  const { dbId, onSuccess, onError } = options || {};
+  const { dbId, catalog, onSuccess, onError } = options || {};
   const [trigger] = useLazySchemasQuery();
   const result = useSchemasQuery(
-    { dbId, forceRefresh: false },
+    { dbId, catalog: catalog || undefined, forceRefresh: false },
     {
       skip: !dbId,
     },
@@ -89,9 +91,24 @@ export function useSchemas(options: Params) {
     onError?.();
   });
 
+  useEffect(() => {
+    if (dbId) {
+      trigger({ dbId, catalog, forceRefresh: false }).then(
+        ({ isSuccess, isError, data }) => {
+          if (isSuccess) {
+            handleOnSuccess(data || EMPTY_SCHEMAS, true);
+          }
+          if (isError) {
+            handleOnError();
+          }
+        },
+      );
+    }
+  }, [dbId, catalog, handleOnError, handleOnSuccess, trigger]);
+
   const refetch = useCallback(() => {
     if (dbId) {
-      trigger({ dbId, forceRefresh: true }).then(
+      trigger({ dbId, catalog, forceRefresh: true }).then(
         ({ isSuccess, isError, data }) => {
           if (isSuccess) {
             handleOnSuccess(data || EMPTY_SCHEMAS, true);
@@ -102,7 +119,7 @@ export function useSchemas(options: Params) {
         },
       );
     }
-  }, [dbId, handleOnError, handleOnSuccess, trigger]);
+  }, [dbId, catalog, handleOnError, handleOnSuccess, trigger]);
 
   useEffect(() => {
     if (isMountedRef.current) {
@@ -119,7 +136,7 @@ export function useSchemas(options: Params) {
     } else {
       isMountedRef.current = true;
     }
-  }, [result, handleOnSuccess, handleOnError]);
+  }, [catalog, result, handleOnSuccess, handleOnError]);
 
   return {
     ...result,
diff --git a/superset-frontend/src/hooks/apiResources/sqlEditorTabs.ts b/superset-frontend/src/hooks/apiResources/sqlEditorTabs.ts
index 71e0cf2936e5a..f25e9b4021aee 100644
--- a/superset-frontend/src/hooks/apiResources/sqlEditorTabs.ts
+++ b/superset-frontend/src/hooks/apiResources/sqlEditorTabs.ts
@@ -33,6 +33,7 @@ const sqlEditorApi = api.injectEndpoints({
           version = LatestQueryEditorVersion,
           id,
           dbId,
+          catalog,
           schema,
           queryLimit,
           sql,
@@ -50,6 +51,7 @@ const sqlEditorApi = api.injectEndpoints({
         postPayload: pickBy(
           {
             database_id: dbId,
+            catalog,
             schema,
             sql,
             label: name,
diff --git a/superset-frontend/src/hooks/apiResources/sqlLab.ts b/superset-frontend/src/hooks/apiResources/sqlLab.ts
index 16e8ffde6c609..45f7c83a38083 100644
--- a/superset-frontend/src/hooks/apiResources/sqlLab.ts
+++ b/superset-frontend/src/hooks/apiResources/sqlLab.ts
@@ -27,6 +27,7 @@ export type InitialState = {
     label: string;
     active: boolean;
     database_id: number;
+    catalog?: string | null;
     schema?: string;
     table_schemas: {
       id: number;
@@ -38,6 +39,7 @@ export type InitialState = {
         }[];
         dataPreviewQueryId?: string;
       } & Record<string, any>;
+      catalog?: string | null;
       schema?: string;
       tab_state_id: number;
       database_id?: number;
diff --git a/superset-frontend/src/hooks/apiResources/tables.test.ts b/superset-frontend/src/hooks/apiResources/tables.test.ts
index e461b4f86b2b5..7919370817db2 100644
--- a/superset-frontend/src/hooks/apiResources/tables.test.ts
+++ b/superset-frontend/src/hooks/apiResources/tables.test.ts
@@ -81,9 +81,11 @@ describe('useTables hook', () => {
   test('returns api response mapping json options', async () => {
     const expectDbId = 'db1';
     const expectedSchema = 'schema1';
+    const catalogApiRoute = `glob:*/api/v1/database/${expectDbId}/catalogs/*`;
     const schemaApiRoute = `glob:*/api/v1/database/${expectDbId}/schemas/*`;
     const tableApiRoute = `glob:*/api/v1/database/${expectDbId}/tables/?q=*`;
     fetchMock.get(tableApiRoute, fakeApiResult);
+    fetchMock.get(catalogApiRoute, { count: 0, result: [] });
     fetchMock.get(schemaApiRoute, {
       result: fakeSchemaApiResult,
     });
@@ -130,9 +132,11 @@ describe('useTables hook', () => {
   test('skips the deprecated schema option', async () => {
     const expectDbId = 'db1';
     const unexpectedSchema = 'invalid schema';
+    const catalogApiRoute = `glob:*/api/v1/database/${expectDbId}/catalogs/*`;
     const schemaApiRoute = `glob:*/api/v1/database/${expectDbId}/schemas/*`;
     const tableApiRoute = `glob:*/api/v1/database/${expectDbId}/tables/?q=*`;
     fetchMock.get(tableApiRoute, fakeApiResult);
+    fetchMock.get(catalogApiRoute, { count: 0, result: [] });
     fetchMock.get(schemaApiRoute, {
       result: fakeSchemaApiResult,
     });
@@ -166,6 +170,10 @@ describe('useTables hook', () => {
     const expectedSchema = 'schema2';
     const tableApiRoute = `glob:*/api/v1/database/${expectDbId}/tables/?q=*`;
     fetchMock.get(tableApiRoute, fakeHasMoreApiResult);
+    fetchMock.get(`glob:*/api/v1/database/${expectDbId}/catalogs/*`, {
+      count: 0,
+      result: [],
+    });
     fetchMock.get(`glob:*/api/v1/database/${expectDbId}/schemas/*`, {
       result: fakeSchemaApiResult,
     });
@@ -191,6 +199,10 @@ describe('useTables hook', () => {
     const expectedSchema = 'schema1';
     const tableApiRoute = `glob:*/api/v1/database/${expectDbId}/tables/?q=*`;
     fetchMock.get(tableApiRoute, fakeApiResult);
+    fetchMock.get(`glob:*/api/v1/database/${expectDbId}/catalogs/*`, {
+      count: 0,
+      result: [],
+    });
     fetchMock.get(`glob:*/api/v1/database/${expectDbId}/schemas/*`, {
       result: fakeSchemaApiResult,
     });
@@ -220,6 +232,10 @@ describe('useTables hook', () => {
     fetchMock.get(tableApiRoute, url =>
       url.includes(expectedSchema) ? fakeApiResult : fakeHasMoreApiResult,
     );
+    fetchMock.get(`glob:*/api/v1/database/${expectDbId}/catalogs/*`, {
+      count: 0,
+      result: [],
+    });
     fetchMock.get(`glob:*/api/v1/database/${expectDbId}/schemas/*`, {
       result: fakeSchemaApiResult,
     });
diff --git a/superset-frontend/src/hooks/apiResources/tables.ts b/superset-frontend/src/hooks/apiResources/tables.ts
index 41be4c167c9c8..d90c528b40aa3 100644
--- a/superset-frontend/src/hooks/apiResources/tables.ts
+++ b/superset-frontend/src/hooks/apiResources/tables.ts
@@ -18,6 +18,7 @@
  */
 import { useCallback, useMemo, useEffect, useRef } from 'react';
 import useEffectEvent from 'src/hooks/useEffectEvent';
+import { toQueryString } from 'src/utils/urlUtils';
 import { api, JsonResponse } from './queryApi';
 
 import { useSchemas } from './schemas';
@@ -50,6 +51,7 @@ export type Data = {
 
 export type FetchTablesQueryParams = {
   dbId?: string | number;
+  catalog?: string | null;
   schema?: string;
   forceRefresh?: boolean;
   onSuccess?: (data: Data, isRefetched: boolean) => void;
@@ -58,6 +60,7 @@ export type FetchTablesQueryParams = {
 
 export type FetchTableMetadataQueryParams = {
   dbId: string | number;
+  catalog?: string | null;
   schema: string;
   table: string;
 };
@@ -95,12 +98,13 @@ const tableApi = api.injectEndpoints({
   endpoints: builder => ({
     tables: builder.query<Data, FetchTablesQueryParams>({
       providesTags: ['Tables'],
-      query: ({ dbId, schema, forceRefresh }) => ({
+      query: ({ dbId, catalog, schema, forceRefresh }) => ({
         endpoint: `/api/v1/database/${dbId ?? 'undefined'}/tables/`,
         // TODO: Would be nice to add pagination in a follow-up. Needs endpoint changes.
         urlParams: {
           force: forceRefresh,
           schema_name: schema ? encodeURIComponent(schema) : '',
+          ...(catalog && { catalog_name: catalog }),
         },
         transformResponse: ({ json }: QueryResponse) => ({
           options: json.result,
@@ -113,10 +117,12 @@ const tableApi = api.injectEndpoints({
       }),
     }),
     tableMetadata: builder.query<TableMetaData, FetchTableMetadataQueryParams>({
-      query: ({ dbId, schema, table }) => ({
-        endpoint: schema
-          ? `/api/v1/database/${dbId}/table_metadata/?name=${table}&schema=${schema}`
-          : `/api/v1/database/${dbId}/table_metadata/?name=${table}`,
+      query: ({ dbId, catalog, schema, table }) => ({
+        endpoint: `/api/v1/database/${dbId}/table_metadata/${toQueryString({
+          name: table,
+          catalog,
+          schema,
+        })}`,
         transformResponse: ({ json }: TableMetadataReponse) => json,
       }),
     }),
@@ -124,10 +130,10 @@ const tableApi = api.injectEndpoints({
       TableExtendedMetadata,
       FetchTableMetadataQueryParams
     >({
-      query: ({ dbId, schema, table }) => ({
-        endpoint: schema
-          ? `/api/v1/database/${dbId}/table_metadata/extra/?name=${table}&schema=${schema}`
-          : `/api/v1/database/${dbId}/table_metadata/extra/?name=${table}`,
+      query: ({ dbId, catalog, schema, table }) => ({
+        endpoint: `/api/v1/database/${dbId}/table_metadata/extra/${toQueryString(
+          { name: table, catalog, schema },
+        )}`,
         transformResponse: ({ json }: JsonResponse) => json,
       }),
     }),
@@ -144,22 +150,23 @@ export const {
 } = tableApi;
 
 export function useTables(options: Params) {
+  const { dbId, catalog, schema, onSuccess, onError } = options || {};
   const isMountedRef = useRef(false);
   const { data: schemaOptions, isFetching } = useSchemas({
-    dbId: options.dbId,
+    dbId,
+    catalog: catalog || undefined,
   });
   const schemaOptionsMap = useMemo(
     () => new Set(schemaOptions?.map(({ value }) => value)),
     [schemaOptions],
   );
-  const { dbId, schema, onSuccess, onError } = options || {};
 
   const enabled = Boolean(
     dbId && schema && !isFetching && schemaOptionsMap.has(schema),
   );
 
   const result = useTablesQuery(
-    { dbId, schema, forceRefresh: false },
+    { dbId, catalog, schema, forceRefresh: false },
     {
       skip: !enabled,
     },
@@ -176,7 +183,7 @@ export function useTables(options: Params) {
 
   const refetch = useCallback(() => {
     if (enabled) {
-      trigger({ dbId, schema, forceRefresh: true }).then(
+      trigger({ dbId, catalog, schema, forceRefresh: true }).then(
         ({ isSuccess, isError, data, error }) => {
           if (isSuccess && data) {
             handleOnSuccess(data, true);
@@ -187,7 +194,7 @@ export function useTables(options: Params) {
         },
       );
     }
-  }, [dbId, schema, enabled, handleOnSuccess, handleOnError, trigger]);
+  }, [dbId, catalog, schema, enabled, handleOnSuccess, handleOnError, trigger]);
 
   useEffect(() => {
     if (isMountedRef.current) {
diff --git a/superset-frontend/src/pages/DatasetCreation/index.tsx b/superset-frontend/src/pages/DatasetCreation/index.tsx
index 66acc4fa235e1..47378e828d5e1 100644
--- a/superset-frontend/src/pages/DatasetCreation/index.tsx
+++ b/superset-frontend/src/pages/DatasetCreation/index.tsx
@@ -48,6 +48,14 @@ export function datasetReducer(
       return {
         ...trimmedState,
         ...action.payload,
+        catalog: null,
+        schema: null,
+        table_name: null,
+      };
+    case DatasetActionType.SelectCatalog:
+      return {
+        ...trimmedState,
+        [action.payload.name]: action.payload.value,
         schema: null,
         table_name: null,
       };
@@ -112,6 +120,7 @@ export default function AddDataset() {
     <DatasetPanel
       tableName={dataset?.table_name}
       dbId={dataset?.db?.id}
+      catalog={dataset?.catalog}
       schema={dataset?.schema}
       setHasColumns={setHasColumns}
       datasets={datasets}
diff --git a/superset-frontend/src/types/Database.ts b/superset-frontend/src/types/Database.ts
index 69e86e3a7b06e..a088fe26d3911 100644
--- a/superset-frontend/src/types/Database.ts
+++ b/superset-frontend/src/types/Database.ts
@@ -29,4 +29,5 @@ export default interface Database {
   catalog: object;
   parameters: any;
   disable_drill_to_detail?: boolean;
+  allow_multi_catalog?: boolean;
 }
diff --git a/superset-frontend/src/utils/datasourceUtils.js b/superset-frontend/src/utils/datasourceUtils.js
index 1a5924b3e6c49..ef984044d1347 100644
--- a/superset-frontend/src/utils/datasourceUtils.js
+++ b/superset-frontend/src/utils/datasourceUtils.js
@@ -21,5 +21,6 @@ export const getDatasourceAsSaveableDataset = source => ({
   name: source?.datasource_name || source?.name || 'Untitled',
   dbId: source?.database?.id || source?.dbId,
   sql: source?.sql || '',
+  catalog: source?.catalog,
   schema: source?.schema,
 });
diff --git a/superset-frontend/src/utils/urlUtils.test.ts b/superset-frontend/src/utils/urlUtils.test.ts
index 19e2a470d1821..d992058821ae2 100644
--- a/superset-frontend/src/utils/urlUtils.test.ts
+++ b/superset-frontend/src/utils/urlUtils.test.ts
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-import { isUrlExternal, parseUrl } from './urlUtils';
+import { isUrlExternal, parseUrl, toQueryString } from './urlUtils';
 
 test('isUrlExternal', () => {
   expect(isUrlExternal('http://google.com')).toBeTruthy();
@@ -52,3 +52,47 @@ test('parseUrl', () => {
   expect(parseUrl('/about')).toEqual('/about');
   expect(parseUrl('#anchor')).toEqual('#anchor');
 });
+
+describe('toQueryString', () => {
+  it('should return an empty string if the input is an empty object', () => {
+    expect(toQueryString({})).toBe('');
+  });
+
+  it('should correctly convert a single key-value pair to a query string', () => {
+    expect(toQueryString({ key: 'value' })).toBe('?key=value');
+  });
+
+  it('should correctly convert multiple key-value pairs to a query string', () => {
+    expect(toQueryString({ key1: 'value1', key2: 'value2' })).toBe(
+      '?key1=value1&key2=value2',
+    );
+  });
+
+  it('should encode URI components', () => {
+    expect(
+      toQueryString({ 'a key': 'a value', email: 'test@example.com' }),
+    ).toBe('?a%20key=a%20value&email=test%40example.com');
+  });
+
+  it('should omit keys with undefined values', () => {
+    expect(toQueryString({ key1: 'value1', key2: undefined })).toBe(
+      '?key1=value1',
+    );
+  });
+
+  it('should omit keys with null values', () => {
+    expect(toQueryString({ key1: 'value1', key2: null })).toBe('?key1=value1');
+  });
+
+  it('should handle numbers and boolean values as parameter values', () => {
+    expect(toQueryString({ number: 123, truth: true, lie: false })).toBe(
+      '?number=123&truth=true&lie=false',
+    );
+  });
+
+  it('should handle special characters in keys and values', () => {
+    expect(toQueryString({ 'user@domain': 'me&you' })).toBe(
+      '?user%40domain=me%26you',
+    );
+  });
+});
diff --git a/superset-frontend/src/utils/urlUtils.ts b/superset-frontend/src/utils/urlUtils.ts
index a44c3ed7a6851..2858d65a7de1c 100644
--- a/superset-frontend/src/utils/urlUtils.ts
+++ b/superset-frontend/src/utils/urlUtils.ts
@@ -206,3 +206,16 @@ export function parseUrl(url: string) {
   }
   return url;
 }
+
+export function toQueryString(params: Record<string, any>): string {
+  const queryParts: string[] = [];
+  Object.keys(params).forEach(key => {
+    const value = params[key];
+    if (value !== null && value !== undefined) {
+      queryParts.push(
+        `${encodeURIComponent(key)}=${encodeURIComponent(value)}`,
+      );
+    }
+  });
+  return queryParts.length > 0 ? `?${queryParts.join('&')}` : '';
+}
diff --git a/superset/cachekeys/api.py b/superset/cachekeys/api.py
index 3c90dafb72723..91cae29b8dc5a 100644
--- a/superset/cachekeys/api.py
+++ b/superset/cachekeys/api.py
@@ -85,6 +85,7 @@ def invalidate(self) -> Response:
         for ds in datasources.get("datasources", []):
             ds_obj = SqlaTable.get_datasource_by_name(
                 datasource_name=ds.get("datasource_name"),
+                catalog=ds.get("catalog"),
                 schema=ds.get("schema"),
                 database_name=ds.get("database_name"),
             )
diff --git a/superset/commands/dashboard/importers/v0.py b/superset/commands/dashboard/importers/v0.py
index 45089cbeac1af..48dcb230ef9a4 100644
--- a/superset/commands/dashboard/importers/v0.py
+++ b/superset/commands/dashboard/importers/v0.py
@@ -66,6 +66,7 @@ def import_chart(
     datasource = SqlaTable.get_datasource_by_name(
         datasource_name=params["datasource_name"],
         database_name=params["database_name"],
+        catalog=params.get("catalog"),
         schema=params["schema"],
     )
     slc_to_import.datasource_id = datasource.id  # type: ignore
diff --git a/superset/commands/database/validate_sql.py b/superset/commands/database/validate_sql.py
index 6a93a01473acb..c13d4d5d33875 100644
--- a/superset/commands/database/validate_sql.py
+++ b/superset/commands/database/validate_sql.py
@@ -60,8 +60,8 @@ def run(self) -> list[dict[str, Any]]:
         if not self._validator or not self._model:
             raise ValidatorSQLUnexpectedError()
         sql = self._properties["sql"]
-        schema = self._properties.get("schema")
         catalog = self._properties.get("catalog")
+        schema = self._properties.get("schema")
         try:
             timeout = current_app.config["SQLLAB_VALIDATION_TIMEOUT"]
             timeout_msg = f"The query exceeded the {timeout} seconds timeout."
diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py
index 12fbdc3bd7baf..ca1adba0accac 100644
--- a/superset/connectors/sqla/models.py
+++ b/superset/connectors/sqla/models.py
@@ -698,7 +698,11 @@ def raise_for_access(self) -> None:
 
     @classmethod
     def get_datasource_by_name(
-        cls, datasource_name: str, schema: str, database_name: str
+        cls,
+        datasource_name: str,
+        catalog: str | None,
+        schema: str,
+        database_name: str,
     ) -> BaseDatasource | None:
         raise NotImplementedError()
 
@@ -1239,6 +1243,7 @@ def database_name(self) -> str:
     def get_datasource_by_name(
         cls,
         datasource_name: str,
+        catalog: str | None,
         schema: str | None,
         database_name: str,
     ) -> SqlaTable | None:
@@ -1248,6 +1253,7 @@ def get_datasource_by_name(
             .join(Database)
             .filter(cls.table_name == datasource_name)
             .filter(Database.database_name == database_name)
+            .filter(cls.catalog == catalog)
         )
         # Handling schema being '' or None, which is easier to handle
         # in python than in the SQLA query in a multi-dialect way
@@ -1752,7 +1758,7 @@ def assign_column_label(df: pd.DataFrame) -> pd.DataFrame | None:
         try:
             df = self.database.get_df(
                 sql,
-                None,
+                self.catalog,
                 self.schema or None,
                 mutator=assign_column_label,
             )
diff --git a/superset/dashboards/schemas.py b/superset/dashboards/schemas.py
index 83c533b86ec9a..29b13ae7b860f 100644
--- a/superset/dashboards/schemas.py
+++ b/superset/dashboards/schemas.py
@@ -216,6 +216,7 @@ class DatabaseSchema(Schema):
     allows_virtual_table_explore = fields.Bool()
     disable_data_preview = fields.Bool()
     disable_drill_to_detail = fields.Bool()
+    allow_multi_catalog = fields.Bool()
     explore_database_id = fields.Int()
 
 
diff --git a/superset/databases/api.py b/superset/databases/api.py
index 116ac9f46b49a..31db3ceacce85 100644
--- a/superset/databases/api.py
+++ b/superset/databases/api.py
@@ -217,6 +217,7 @@ class DatabaseRestApi(BaseSupersetModelRestApi):
         "uuid",
         "disable_data_preview",
         "disable_drill_to_detail",
+        "allow_multi_catalog",
         "engine_information",
     ]
     add_columns = [
diff --git a/superset/databases/schemas.py b/superset/databases/schemas.py
index 4318a1b48c7a0..e6f6eaa12e37a 100644
--- a/superset/databases/schemas.py
+++ b/superset/databases/schemas.py
@@ -156,7 +156,9 @@
     "6. The ``disable_data_preview`` field is a boolean specifying whether or not data "
     "preview queries will be run when fetching table metadata in SQL Lab."
     "7. The ``disable_drill_to_detail`` field is a boolean specifying whether or not"
-    "drill to detail is disabled for the database.",
+    "drill to detail is disabled for the database."
+    "8. The ``allow_multi_catalog`` indicates if the database allows changing "
+    "the default catalog when running queries and creating datasets.",
     True,
 )
 get_export_ids_schema = {"type": "array", "items": {"type": "integer"}}
@@ -739,6 +741,7 @@ class ValidateSQLRequest(Schema):
     sql = fields.String(
         required=True, metadata={"description": "SQL statement to validate"}
     )
+    catalog = fields.String(required=False, allow_none=True)
     schema = fields.String(required=False, allow_none=True)
     template_params = fields.Dict(required=False, allow_none=True)
 
@@ -824,6 +827,7 @@ def fix_schemas_allowed_for_csv_upload(  # pylint: disable=invalid-name
     cancel_query_on_windows_unload = fields.Boolean(required=False)
     disable_data_preview = fields.Boolean(required=False)
     disable_drill_to_detail = fields.Boolean(required=False)
+    allow_multi_catalog = fields.Boolean(required=False)
     version = fields.String(required=False, allow_none=True)
 
 
@@ -968,6 +972,20 @@ class DatabaseSchemaAccessForFileUploadResponse(Schema):
     )
 
 
+class EngineInformationSchema(Schema):
+    supports_file_upload = fields.Boolean(
+        metadata={"description": "Users can upload files to the database"}
+    )
+    disable_ssh_tunneling = fields.Boolean(
+        metadata={"description": "SSH tunnel is not available to the database"}
+    )
+    supports_dynamic_catalog = fields.Boolean(
+        metadata={
+            "description": "The database supports multiple catalogs in a single connection"
+        }
+    )
+
+
 class DatabaseConnectionSchema(Schema):
     """
     Schema with database connection information.
@@ -1001,7 +1019,7 @@ class DatabaseConnectionSchema(Schema):
     driver = fields.String(
         allow_none=True, metadata={"description": "SQLAlchemy driver to use"}
     )
-    engine_information = fields.Dict(keys=fields.String(), values=fields.Raw())
+    engine_information = fields.Nested(EngineInformationSchema)
     expose_in_sqllab = fields.Boolean(
         metadata={"description": expose_in_sqllab_description}
     )
diff --git a/superset/datasets/api.py b/superset/datasets/api.py
index f6dedc97ebec3..0b793392d734c 100644
--- a/superset/datasets/api.py
+++ b/superset/datasets/api.py
@@ -119,6 +119,7 @@ class DatasetRestApi(BaseSupersetModelRestApi):
         "owners.id",
         "owners.first_name",
         "owners.last_name",
+        "catalog",
         "schema",
         "sql",
         "table_name",
@@ -126,6 +127,7 @@ class DatasetRestApi(BaseSupersetModelRestApi):
     list_select_columns = list_columns + ["changed_on", "changed_by_fk"]
     order_columns = [
         "table_name",
+        "catalog",
         "schema",
         "changed_by.first_name",
         "changed_on_delta_humanized",
@@ -139,6 +141,7 @@ class DatasetRestApi(BaseSupersetModelRestApi):
         "sql",
         "filter_select_enabled",
         "fetch_values_predicate",
+        "catalog",
         "schema",
         "description",
         "main_dttm_col",
@@ -197,6 +200,7 @@ class DatasetRestApi(BaseSupersetModelRestApi):
     show_columns = show_select_columns + [
         "columns.type_generic",
         "database.backend",
+        "database.allow_multi_catalog",
         "columns.advanced_data_type",
         "is_managed_externally",
         "uid",
@@ -212,12 +216,13 @@ class DatasetRestApi(BaseSupersetModelRestApi):
     add_model_schema = DatasetPostSchema()
     edit_model_schema = DatasetPutSchema()
     duplicate_model_schema = DatasetDuplicateSchema()
-    add_columns = ["database", "schema", "table_name", "sql", "owners"]
+    add_columns = ["database", "catalog", "schema", "table_name", "sql", "owners"]
     edit_columns = [
         "table_name",
         "sql",
         "filter_select_enabled",
         "fetch_values_predicate",
+        "catalog",
         "schema",
         "description",
         "main_dttm_col",
@@ -251,6 +256,7 @@ class DatasetRestApi(BaseSupersetModelRestApi):
         "id",
         "database",
         "owners",
+        "catalog",
         "schema",
         "sql",
         "table_name",
@@ -258,7 +264,7 @@ class DatasetRestApi(BaseSupersetModelRestApi):
         "changed_by",
     ]
     allowed_rel_fields = {"database", "owners", "created_by", "changed_by"}
-    allowed_distinct_fields = {"schema"}
+    allowed_distinct_fields = {"catalog", "schema"}
 
     apispec_parameter_schemas = {
         "get_export_ids_schema": get_export_ids_schema,
diff --git a/superset/datasets/schemas.py b/superset/datasets/schemas.py
index 45a44043a5d23..5ce0621675958 100644
--- a/superset/datasets/schemas.py
+++ b/superset/datasets/schemas.py
@@ -93,6 +93,7 @@ class DatasetMetricsPutSchema(Schema):
 
 class DatasetPostSchema(Schema):
     database = fields.Integer(required=True)
+    catalog = fields.String(allow_none=True, validate=Length(0, 250))
     schema = fields.String(allow_none=True, validate=Length(0, 250))
     table_name = fields.String(required=True, allow_none=False, validate=Length(1, 250))
     sql = fields.String(allow_none=True)
@@ -109,6 +110,7 @@ class DatasetPutSchema(Schema):
     sql = fields.String(allow_none=True)
     filter_select_enabled = fields.Boolean(allow_none=True)
     fetch_values_predicate = fields.String(allow_none=True, validate=Length(0, 1000))
+    catalog = fields.String(allow_none=True, validate=Length(0, 250))
     schema = fields.String(allow_none=True, validate=Length(0, 255))
     description = fields.String(allow_none=True)
     main_dttm_col = fields.String(allow_none=True)
@@ -272,6 +274,11 @@ class GetOrCreateDatasetSchema(Schema):
     database_id = fields.Integer(
         required=True, metadata={"description": "ID of database table belongs to"}
     )
+    catalog = fields.String(
+        allow_none=True,
+        validate=Length(0, 250),
+        metadata={"description": "The catalog the table belongs to"},
+    )
     schema = fields.String(
         allow_none=True,
         validate=Length(0, 250),
diff --git a/superset/db_engine_specs/base.py b/superset/db_engine_specs/base.py
index 5b8a1deaba085..bcee08708d632 100644
--- a/superset/db_engine_specs/base.py
+++ b/superset/db_engine_specs/base.py
@@ -2184,6 +2184,7 @@ def get_public_information(cls) -> dict[str, Any]:
         return {
             "supports_file_upload": cls.supports_file_upload,
             "disable_ssh_tunneling": cls.disable_ssh_tunneling,
+            "supports_dynamic_catalog": cls.supports_dynamic_catalog,
         }
 
     @classmethod
diff --git a/superset/models/core.py b/superset/models/core.py
index fe486bf2b1992..bbf30523420a6 100755
--- a/superset/models/core.py
+++ b/superset/models/core.py
@@ -235,6 +235,10 @@ def disable_drill_to_detail(self) -> bool:
         # this will prevent any 'trash value' strings from going through
         return self.get_extra().get("disable_drill_to_detail", False) is True
 
+    @property
+    def allow_multi_catalog(self) -> bool:
+        return self.get_extra().get("allow_multi_catalog", False)
+
     @property
     def schema_options(self) -> dict[str, Any]:
         """Additional schema display config for engines with complex schemas"""
@@ -255,6 +259,7 @@ def data(self) -> dict[str, Any]:
             "parameters": self.parameters,
             "disable_data_preview": self.disable_data_preview,
             "disable_drill_to_detail": self.disable_drill_to_detail,
+            "allow_multi_catalog": self.allow_multi_catalog,
             "parameters_schema": self.parameters_schema,
             "engine_information": self.engine_information,
         }
diff --git a/superset/models/sql_lab.py b/superset/models/sql_lab.py
index 31443b4bb13e6..4c06867501bbd 100644
--- a/superset/models/sql_lab.py
+++ b/superset/models/sql_lab.py
@@ -514,6 +514,7 @@ def to_dict(self) -> dict[str, Any]:
             "label": self.label,
             "active": self.active,
             "database_id": self.database_id,
+            "catalog": self.catalog,
             "schema": self.schema,
             "table_schemas": [ts.to_dict() for ts in self.table_schemas],
             "sql": self.sql,
diff --git a/superset/queries/saved_queries/api.py b/superset/queries/saved_queries/api.py
index be9a3f00b35a7..d772483d9c4d8 100644
--- a/superset/queries/saved_queries/api.py
+++ b/superset/queries/saved_queries/api.py
@@ -95,6 +95,7 @@ class SavedQueryRestApi(BaseSupersetModelRestApi):
         "description",
         "id",
         "label",
+        "catalog",
         "schema",
         "sql",
         "sql_tables",
@@ -119,6 +120,7 @@ class SavedQueryRestApi(BaseSupersetModelRestApi):
         "label",
         "last_run_delta_humanized",
         "rows",
+        "catalog",
         "schema",
         "sql",
         "sql_tables",
@@ -130,12 +132,14 @@ class SavedQueryRestApi(BaseSupersetModelRestApi):
         "db_id",
         "description",
         "label",
+        "catalog",
         "schema",
         "sql",
         "template_parameters",
     ]
     edit_columns = add_columns
     order_columns = [
+        "catalog",
         "schema",
         "label",
         "description",
@@ -148,7 +152,15 @@ class SavedQueryRestApi(BaseSupersetModelRestApi):
         "last_run_delta_humanized",
     ]
 
-    search_columns = ["id", "database", "label", "schema", "created_by", "changed_by"]
+    search_columns = [
+        "id",
+        "database",
+        "label",
+        "catalog",
+        "schema",
+        "created_by",
+        "changed_by",
+    ]
     if is_feature_enabled("TAGGING_SYSTEM"):
         search_columns += ["tags"]
     search_filters = {
@@ -170,7 +182,7 @@ class SavedQueryRestApi(BaseSupersetModelRestApi):
     }
     base_related_field_filters = {"database": [["id", DatabaseFilter, lambda: []]]}
     allowed_rel_fields = {"database", "changed_by", "created_by"}
-    allowed_distinct_fields = {"schema"}
+    allowed_distinct_fields = {"catalog", "schema"}
 
     def pre_add(self, item: SavedQuery) -> None:
         item.user = g.user
diff --git a/superset/sqllab/schemas.py b/superset/sqllab/schemas.py
index 0864420c9077d..5e22a97e2a73c 100644
--- a/superset/sqllab/schemas.py
+++ b/superset/sqllab/schemas.py
@@ -53,6 +53,7 @@ class ExecutePayloadSchema(Schema):
     client_id = fields.String(allow_none=True)
     queryLimit = fields.Integer(allow_none=True)
     sql_editor_id = fields.String(allow_none=True)
+    catalog = fields.String(allow_none=True)
     schema = fields.String(allow_none=True)
     tab = fields.String(allow_none=True)
     ctas_method = fields.String(allow_none=True)
diff --git a/superset/sqllab/sqllab_execution_context.py b/superset/sqllab/sqllab_execution_context.py
index 7a732cf64264d..7ab4459be3675 100644
--- a/superset/sqllab/sqllab_execution_context.py
+++ b/superset/sqllab/sqllab_execution_context.py
@@ -44,6 +44,7 @@
 @dataclass
 class SqlJsonExecutionContext:  # pylint: disable=too-many-instance-attributes
     database_id: int
+    catalog: str | None
     schema: str
     sql: str
     template_params: dict[str, Any]
@@ -73,6 +74,7 @@ def set_query(self, query: Query) -> None:
 
     def _init_from_query_params(self, query_params: dict[str, Any]) -> None:
         self.database_id = cast(int, query_params.get("database_id"))
+        self.catalog = cast(str, query_params.get("catalog"))
         self.schema = cast(str, query_params.get("schema"))
         self.sql = cast(str, query_params.get("sql"))
         self.template_params = self._get_template_params(query_params)
@@ -147,6 +149,7 @@ def create_query(self) -> Query:
             return Query(
                 database_id=self.database_id,
                 sql=self.sql,
+                catalog=self.catalog,
                 schema=self.schema,
                 select_as_cta=True,
                 ctas_method=self.create_table_as_select.ctas_method,  # type: ignore
@@ -163,6 +166,7 @@ def create_query(self) -> Query:
         return Query(
             database_id=self.database_id,
             sql=self.sql,
+            catalog=self.catalog,
             schema=self.schema,
             select_as_cta=False,
             start_time=start_time,
diff --git a/superset/sqllab/utils.py b/superset/sqllab/utils.py
index bbf39196403f9..65b87bbf6e04f 100644
--- a/superset/sqllab/utils.py
+++ b/superset/sqllab/utils.py
@@ -39,6 +39,7 @@
     "id",
     "disable_data_preview",
     "disable_drill_to_detail",
+    "allow_multi_catalog",
 ]
 
 
diff --git a/superset/views/database/mixins.py b/superset/views/database/mixins.py
index 0d104aad5f553..21c664fa1f4ea 100644
--- a/superset/views/database/mixins.py
+++ b/superset/views/database/mixins.py
@@ -149,7 +149,9 @@ class DatabaseMixin:
             "not data preview queries will be run when fetching table metadata in"
             "SQL Lab."
             "7. The ``disable_drill_to_detail`` field is a boolean specifying whether or"
-            "not drill to detail is disabled for the database.",
+            "not drill to detail is disabled for the database."
+            "8. The ``allow_multi_catalog`` indicates if the database allows changing "
+            "the default catalog when running queries and creating datasets.",
             True,
         ),
         "encrypted_extra": utils.markdown(
diff --git a/superset/views/datasource/schemas.py b/superset/views/datasource/schemas.py
index 73496b02b2855..8ae28f9c7a212 100644
--- a/superset/views/datasource/schemas.py
+++ b/superset/views/datasource/schemas.py
@@ -26,6 +26,7 @@
 class ExternalMetadataParams(TypedDict):
     datasource_type: str
     database_name: str
+    catalog_name: Optional[str]
     schema_name: str
     table_name: str
     normalize_columns: Optional[bool]
@@ -45,6 +46,7 @@ class ExternalMetadataParams(TypedDict):
 class ExternalMetadataSchema(Schema):
     datasource_type = fields.Str(required=True)
     database_name = fields.Str(required=True)
+    catalog_name = fields.Str(allow_none=True)
     schema_name = fields.Str(allow_none=True)
     table_name = fields.Str(required=True)
     normalize_columns = fields.Bool(allow_none=True)
@@ -60,6 +62,7 @@ def normalize(
         return ExternalMetadataParams(
             datasource_type=data["datasource_type"],
             database_name=data["database_name"],
+            catalog_name=data.get("catalog_name"),
             schema_name=data.get("schema_name", ""),
             table_name=data["table_name"],
             normalize_columns=data["normalize_columns"],
diff --git a/superset/views/datasource/views.py b/superset/views/datasource/views.py
index 7f81081777538..27371a5d2e188 100644
--- a/superset/views/datasource/views.py
+++ b/superset/views/datasource/views.py
@@ -165,6 +165,7 @@ def external_metadata_by_name(self, **kwargs: Any) -> FlaskResponse:
 
         datasource = SqlaTable.get_datasource_by_name(
             database_name=params["database_name"],
+            catalog=params.get("catalog_name"),
             schema=params["schema_name"],
             datasource_name=params["table_name"],
         )
diff --git a/superset/views/sql_lab/views.py b/superset/views/sql_lab/views.py
index 4ed5143bb651a..6e2738ea2e764 100644
--- a/superset/views/sql_lab/views.py
+++ b/superset/views/sql_lab/views.py
@@ -91,6 +91,7 @@ def post(self) -> FlaskResponse:
             or query_editor.get("title", __("Untitled Query")),
             active=True,
             database_id=query_editor["dbId"],
+            catalog=query_editor.get("catalog"),
             schema=query_editor.get("schema"),
             sql=query_editor.get("sql", "SELECT ..."),
             query_limit=query_editor.get("queryLimit"),