console: add citus support

### Description Add console support for Citus DB ### Changelist - [x] Add/Remove Citus Datasource - [x] Track/Untrack tables - [x] Create Relationships + Tracking suggested relationships - [x] Permissions - [x] Browse Rows (minimal) - [x] Insert Row (using Mutations) - [x] Filtering Shards and Partitions (partition details are not present in Modify tab, it should work once https://github.com/hasura/graphql-engine-mono/pull/1164 is merged) - [x] Create Table - [x] Fixes issue https://github.com/hasura/graphql-engine/issues/6926 for Citus. ### Known Issues - Unable to create Functions from Raw SQL, server returns 400. - Unable to run `ALTER` SQL commands, server returns 400. ### Screenshots ![Screenshot 2021-04-21 at 7 53 23 PM](https://user-images.githubusercontent.com/11921040/115569900-468b3d80-a2db-11eb-8374-e06d5d61b2e4.png) ### Setting up a Citus Source - Link to [docker compose](https://github.com/citusdata/docker/blob/master/docker-compose.yml) - Create an `.envfile` and add the following ```env COORDINATOR_EXTERNAL_PORT=<port to expose citus master> COMPOSE_PROJECT_NAME=test ``` - run `docker-compose --file docker-compose.yml --env-file .envfile up --scale worker=2 -d`. `worker` controls the number of citus worker nodes. - Link to [sample data](http://docs.citusdata.com/en/v10.0/get_started/tutorial_multi_tenant.html#data-model-and-sample-data) ### Changelog - [x] console: add citus support https://github.com/hasura/graphql-engine-mono/pull/1184 Co-authored-by: Vijay Prasanna <11921040+vijayprasanna13@users.noreply.github.com> GitOrigin-RevId: 2500a5d5f996a2904fe8b5c656d6f3f2db707db6
2024-11-10 10:29:12 +03:00 · 2021-06-21 07:46:29 +01:00 · 2021-06-21 07:46:29 +01:00 · 6bb8df6a26
commit 6bb8df6a26
parent acb69dc88c
40 changed files with 2512 additions and 197 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@ -8,6 +8,7 @@
 - server: Adds caching support for queries using remote schema permissions
 - server: All Postgres boolean operators now support the null-collapsing behaviour described in [#704](https://github.com/hasura/graphql-engine/issues/704) and enabled via the `HASURA_GRAPHQL_V1_BOOLEAN_NULL_COLLAPSE` environment variable.
 - cli: `metadata diff` will now only show the differences in metadata. old behaviour is avialble behind a flag (`--type unified-common`) (#5487)
+- console: add citus support

 ## v2.0.0-beta.2

--- a/console/cypress/support/index.d.ts
+++ b/console/cypress/support/index.d.ts
@ -1,5 +1,6 @@
 // type definition for all custom commands
 declare namespace Cypress {
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
  interface Chainable<Subject> {
    /**
     * Custom command to select DOM element by data-test attribute.
--- a/console/jest.config.js
+++ b/console/jest.config.js
@ -3,7 +3,7 @@ module.exports = {
  transform: {
    '^.+\\.(ts|tsx|js|jsx)?$': 'ts-jest',
  },
-  testRegex: '(/__tests__/.*|(\\.|/)(test|spec))\\.tsx?$',
+  testRegex: '(/__tests__/.*)\\.(test|spec).[jt]sx?$',
  moduleFileExtensions: ['ts', 'tsx', 'js', 'jsx', 'json', 'node'],
  moduleDirectories: ['node_modules', 'src'],
  testEnvironment: 'jsdom',
--- a/console/src/components/Common/utils/v1QueryUtils.ts
+++ b/console/src/components/Common/utils/v1QueryUtils.ts
@ -16,7 +16,7 @@ export const getRunSqlQuery = (
  driver = currentDriver
 ) => {
  let type = 'run_sql';
-  if (driver === 'mssql' || driver === 'bigquery') {
+  if (['mssql', 'bigquery', 'citus'].includes(driver)) {
    type = `${driver}_run_sql`;
  }

--- a/console/src/components/Services/Data/Common/getMigrateUrl.ts
+++ b/console/src/components/Services/Data/Common/getMigrateUrl.ts
@ -11,6 +11,7 @@ const rqlQueryTypes = [
  'update',
  'run_sql',
  'mssql_run_sql',
+  'citus_run_sql',
 ];

 type Query = {
--- a/console/src/components/Services/Data/DataActions.js
+++ b/console/src/components/Services/Data/DataActions.js
@ -30,6 +30,7 @@ import {
  mergeDataMssql,
  mergeLoadSchemaDataPostgres,
  mergeDataBigQuery,
+  mergeDataCitus,
 } from './mergeData';
 import _push from './push';
 import { convertArrayToJson } from './TableModify/utils';
@ -231,6 +232,9 @@ const loadSchema = configOptions => {
            case 'postgres':
              mergedData = mergeLoadSchemaDataPostgres(data, metadataTables);
              break;
+            case 'citus':
+              mergedData = mergeDataCitus(data, metadataTables);
+              break;
            case 'mssql':
              mergedData = mergeDataMssql(data, metadataTables);
              break;
--- a/console/src/components/Services/Data/DataSources/ConnectDBForm.tsx
+++ b/console/src/components/Services/Data/DataSources/ConnectDBForm.tsx
@ -42,6 +42,7 @@ export const connectionRadios = [

 const dbTypePlaceholders: Record<Driver, string> = {
  postgres: 'postgresql://username:password@hostname:5432/database',
+  citus: 'postgresql://username:password@hostname:5432/database',
  mssql:
    'Driver={ODBC Driver 17 for SQL Server};Server=serveraddress;Database=dbname;Uid=username;Pwd=password;',
  mysql: 'MySQL connection string',
@ -69,6 +70,10 @@ const driverToLabel: Record<
      'Only Connection Parameters and Environment Variables are available for BigQuery',
    beta: true,
  },
+  citus: {
+    label: 'Citus',
+    defaultConnection: 'DATABASE_URL',
+  },
 };

 const supportedDrivers = getSupportedDrivers('connectDbForm.enabled');
--- a/console/src/components/Services/Data/DataSources/ManageDBUtils.ts
+++ b/console/src/components/Services/Data/DataSources/ManageDBUtils.ts
@ -54,7 +54,7 @@ export const makeConnectionStringFromConnectionParams = ({
  if (password) {
    tPassword = password.trim();
  }
-  if (dbType === 'postgres') {
+  if (dbType === 'postgres' || dbType === 'citus') {
    if (!password) {
      return `postgresql://${tUserName}@${tHost}:${tPort}/${tDatabase}`;
    }
--- a/console/src/components/Services/Data/RawSQL/RawSQL.js
+++ b/console/src/components/Services/Data/RawSQL/RawSQL.js
@ -34,6 +34,13 @@ import { getDataSources } from '../../../../metadata/selector';
 import { services } from '../../../../dataSources/services';
 import { isFeatureSupported, setDriver } from '../../../../dataSources';
 import { fetchDataInit, UPDATE_CURRENT_DATA_SOURCE } from '../DataActions';
+
+const checkChangeLang = (sql, selectedDriver) => {
+  return (
+    !sql?.match(/(?:\$\$\s+)?language\s+plpgsql/i) && selectedDriver === 'citus'
+  );
+};
+
 /**
 * # RawSQL React FC
 * ## renders raw SQL page on route `/data/sql`
@ -82,6 +89,7 @@ const RawSQL = ({

  const [selectedDatabase, setSelectedDatabase] = useState(currentDataSource);
  const [selectedDriver, setSelectedDriver] = useState('postgres');
+  const [suggestLangChange, setSuggestLangChange] = useState(false);

  useEffect(() => {
    const driver = getSourceDriver(sources, selectedDatabase);
@ -115,6 +123,14 @@ const RawSQL = ({
    };
  }, [dispatch, sql, sqlText]);

+  useEffect(() => {
+    if (checkChangeLang(sql, selectedDriver)) {
+      setSuggestLangChange(true);
+    } else {
+      setSuggestLangChange(false);
+    }
+  }, [sql, selectedDriver]);
+
  const submitSQL = () => {
    if (!sqlText) {
      setLSItem(LS_KEYS.rawSQLKey, '');
@ -351,6 +367,7 @@ const RawSQL = ({
              className={`${styles.add_mar_right_small} ${styles.cursorPointer}`}
              id="track-checkbox"
              type="checkbox"
+              disabled={checkChangeLang()}
              onChange={dispatchTrackThis}
              data-test="raw-sql-track-check"
            />
@ -469,7 +486,7 @@ const RawSQL = ({
      </div>
      <div className={styles.add_mar_top}>
        <div className={`${styles.padd_left_remove} col-xs-8`}>
-          <NotesSection />
+          <NotesSection suggestLangChange={suggestLangChange} />
        </div>
        <div className={`${styles.padd_left_remove} col-xs-8`}>
          <label>
--- a/console/src/components/Services/Data/RawSQL/molecules/NotesSection.tsx
+++ b/console/src/components/Services/Data/RawSQL/molecules/NotesSection.tsx
@ -1,6 +1,8 @@
 import React from 'react';

-const NotesSection = () => {
+const NotesSection: React.FC<{
+  suggestLangChange: boolean;
+}> = ({ suggestLangChange }) => {
  return (
    <ul>
      <li>
@ -15,6 +17,19 @@ const NotesSection = () => {
        Multiple SQL statements will be run as a transaction. i.e. if any
        statement fails, none of the statements will be applied.
      </li>
+      {suggestLangChange && (
+        <li>
+          Consider changing custom function language to{' '}
+          <a
+            href="https://www.postgresql.org/docs/13/plpgsql-structure.html"
+            target="_blank"
+            rel="noopener noreferrer"
+          >
+            plpgsql
+          </a>
+          , as citus doesn&apos;t support <code>sql</code>
+        </li>
+      )}
    </ul>
  );
 };
--- a/console/src/components/Services/Data/RawSQL/utils.js
+++ b/console/src/components/Services/Data/RawSQL/utils.js
@ -6,7 +6,7 @@ const getSQLValue = value => {

  let sqlValue = value;
  if (!quotedStringRegex.test(value)) {
-    sqlValue = value.toLowerCase();
+    sqlValue = value?.toLowerCase() ?? '';
  }

  return sqlValue.replace(/['"]+/g, '');
@ -23,7 +23,7 @@ export const removeCommentsSQL = sql => {
 };

 const getDefaultSchema = driver => {
-  if (driver === 'postgres') return 'public';
+  if (driver === 'postgres' || driver === 'citus') return 'public';
  if (driver === 'mssql') return 'dbo';
 };

@ -37,13 +37,13 @@ export const parseCreateSQL = (sql, driver = currentDriver) => {
  const _objects = [];
  const regExp = services[driver].createSQLRegex;
  for (const result of sql.matchAll(regExp)) {
-    const { type, schema, table, tableWithSchema, partition } =
+    const { type, schema, name, nameWithSchema, partition } =
      result.groups ?? {};
-    if (!type || !(table || tableWithSchema)) continue;
+    if (!type || !(name || nameWithSchema)) continue;
    _objects.push({
      type: type.toLowerCase(),
      schema: getSQLValue(schema || getDefaultSchema(driver)),
-      name: getSQLValue(table || tableWithSchema),
+      name: getSQLValue(name || nameWithSchema),
      isPartition: !!partition,
    });
  }
--- a/console/src/components/Services/Data/TableBrowseRows/EditActions.js
+++ b/console/src/components/Services/Data/TableBrowseRows/EditActions.js
@ -15,6 +15,7 @@ import {
 import { getEnumOptionsQuery } from '../../../Common/utils/v1QueryUtils';
 import { isStringArray } from '../../../Common/utils/jsUtils';
 import { generateTableDef } from '../../../../dataSources';
+import { getTableConfiguration } from './utils';

 const E_SET_EDITITEM = 'EditItem/E_SET_EDITITEM';
 const E_ONGOING_REQ = 'EditItem/E_ONGOING_REQ';
@ -31,10 +32,11 @@ const modalClose = () => ({ type: MODAL_CLOSE });
 /* ****************** edit action creators ************ */
 const editItem = (tableName, colValues) => {
  return (dispatch, getState) => {
-    const state = getState();
-
+    const { tables, metadata } = getState();
+    const sources = metadata.metadataObject?.sources;
+    const tableConfiguration = getTableConfiguration(tables, sources);
    /* Type all the values correctly */
-    const { currentSchema, allSchemas, currentDataSource } = state.tables;
+    const { currentSchema, allSchemas, currentDataSource } = tables;

    const tableDef = generateTableDef(tableName, currentSchema);

@ -108,23 +110,29 @@ const editItem = (tableName, colValues) => {
      });
    }

-    const reqBody = {
-      type: 'update',
-      args: {
-        source: currentDataSource,
-        table: tableDef,
-        $set: _setObject,
-        $default: _defaultArray,
-        where: state.tables.update.pkClause,
-      },
-    };
+    if (!dataSource.generateEditRowRequest) return;
+
+    const {
+      getEditRowRequestBody,
+      processEditData,
+      endpoint: url,
+    } = dataSource.generateEditRowRequest();
+
+    const reqBody = getEditRowRequestBody({
+      source: currentDataSource,
+      tableDef,
+      tableConfiguration,
+      set: _setObject,
+      defaultArray: _defaultArray,
+      where: tables.update.pkClause,
+    });
+
    const options = {
      method: 'POST',
      credentials: globalCookiePolicy,
      headers: dataHeaders(getState),
      body: JSON.stringify(reqBody),
    };
-    const url = Endpoints.query;

    return dispatch(
      requestAction(url, options, E_REQUEST_SUCCESS, E_REQUEST_ERROR)
@ -133,7 +141,8 @@ const editItem = (tableName, colValues) => {
        dispatch(
          showSuccessNotification(
            'Edited!',
-            'Affected rows: ' + data.affected_rows
+            'Affected rows: ' +
+              processEditData({ data, tableDef, tableConfiguration })
          )
        );
      },
--- a/console/src/components/Services/Data/TableBrowseRows/EditItem.js
+++ b/console/src/components/Services/Data/TableBrowseRows/EditItem.js
@ -14,6 +14,7 @@ import { getTableBrowseRoute } from '../../../Common/utils/routesUtils';
 import { fetchEnumOptions } from './EditActions';
 import { TableRow } from '../Common/Components/TableRow';
 import { RightContainer } from '../../../Common/Layout/RightContainer';
+import styles from '../../../Common/TableCommon/Table.scss';

 class EditItem extends Component {
  constructor() {
@ -57,8 +58,6 @@ class EditItem extends Component {
      return null;
    }

-    const styles = require('../../../Common/TableCommon/Table.scss');
-
    const currentTable = findTable(
      schemas,
      generateTableDef(tableName, currentSchema)
--- a/console/src/components/Services/Data/TableBrowseRows/ViewActions.js
+++ b/console/src/components/Services/Data/TableBrowseRows/ViewActions.js
@ -1,5 +1,5 @@
 import { defaultViewState } from '../DataState';
-import Endpoints, { globalCookiePolicy } from '../../../../Endpoints';
+import { globalCookiePolicy } from '../../../../Endpoints';
 import requestAction from 'utils/requestAction';
 import filterReducer from './FilterActions';
 import {
@ -8,19 +8,14 @@ import {
 } from '../../Common/Notification';
 import dataHeaders from '../Common/Headers';
 import { getConfirmation } from '../../../Common/utils/jsUtils';
-import {
-  getBulkDeleteQuery,
-  getSelectQuery,
-  getDeleteQuery,
-  getRunSqlQuery,
-} from '../../../Common/utils/v1QueryUtils';
+import { isEmpty } from '../../../Common/utils/jsUtils';
 import { COUNT_LIMIT } from '../constants';
 import {
-  generateTableDef,
  dataSource,
  findTableFromRel,
  isFeatureSupported,
 } from '../../../../dataSources';
+import { getTableConfiguration } from './utils';

 /* ****************** View actions *************/
 const V_SET_DEFAULTS = 'ViewTable/V_SET_DEFAULTS';
@ -53,6 +48,9 @@ const vCollapseRow = () => ({

 const vSetDefaults = limit => ({ type: V_SET_DEFAULTS, limit });

+const showError = (err, msg, dispatch) =>
+  dispatch(showErrorNotification(msg, err.error, err));
+
 const getConfiguration = (tables, sources) => {
  const {
    currentSchema,
@ -70,7 +68,7 @@ const vMakeRowsRequest = () => {
  return (dispatch, getState) => {
    const { tables, metadata } = getState();
    const sources = metadata.metadataObject?.sources;
-    const tableConfiguration = getConfiguration(tables, sources);
+    const tableConfiguration = getTableConfiguration(tables, sources);
    const headers = dataHeaders(getState);
    dispatch({ type: V_REQUEST_PROGRESS, data: true });

@ -156,7 +154,7 @@ const vMakeExportRequest = () => {

 const vMakeCountRequest = () => {
  // For datasources that do not supported aggregation like count
-  if (!isFeatureSupported('tables.browse.aggregation'))
+  if (!isFeatureSupported('tables.browse.aggregation')) {
    return (dispatch, getState) => {
      const { estimatedCount } = getState().tables.view;
      dispatch({
@ -164,40 +162,20 @@ const vMakeCountRequest = () => {
        count: estimatedCount,
      });
    };
+  }
  return (dispatch, getState) => {
+    if (!dataSource.generateRowsCountRequest) return;
+
+    const { tables, metadata } = getState();
+    const sources = metadata.metadataObject?.sources;
+    const tableConfiguration = getConfiguration(tables, sources);
+
    const {
-      currentTable: originalTable,
-      currentSchema,
-      view,
-      currentDataSource,
-    } = getState().tables;
-    const url = Endpoints.query;
-
-    const selectQuery = getSelectQuery(
-      'count',
-      generateTableDef(originalTable, currentSchema),
-      view.query.columns,
-      view.query.where,
-      view.query.offset,
-      view.query.limit,
-      view.query.order_by,
-      currentDataSource
-    );
-
-    let queries = [selectQuery];
-
-    if (dataSource.getStatementTimeoutSql) {
-      queries = [
-        getRunSqlQuery(dataSource.getStatementTimeoutSql(2), currentDataSource),
-        ...queries,
-      ];
-    }
-
-    const requestBody = {
-      type: 'bulk',
-      source: currentDataSource,
-      args: queries,
-    };
+      endpoint,
+      getRowsCountRequestBody,
+      processCount,
+    } = dataSource.generateRowsCountRequest();
+    const requestBody = getRowsCountRequestBody({ tables, tableConfiguration });

    const options = {
      method: 'POST',
@ -206,16 +184,22 @@ const vMakeCountRequest = () => {
      credentials: globalCookiePolicy,
    };

-    return dispatch(requestAction(url, options)).then(
+    return dispatch(requestAction(endpoint, options)).then(
      data => {
-        if (data.length > 1) {
+        if (!isEmpty(data)) {
+          const { currentTable: originalTable, currentSchema } = tables;
+          const count = processCount({
+            data,
+            currentSchema,
+            originalTable,
+            tableConfiguration,
+          });
          const currentTable = getState().tables.currentTable;
-
          // in case table has changed before count load
          if (currentTable === originalTable) {
            dispatch({
              type: V_COUNT_REQUEST_SUCCESS,
-              count: data[1].count,
+              count,
            });
          }
        }
@ -260,10 +244,34 @@ const deleteItem = (pkClause, tableName, tableSchema) => {
    }

    const source = getState().tables.currentDataSource;
+    const { tables, metadata } = getState();
+    const { currentTable: originalTable, allSchemas, currentSchema } = tables;

-    const url = Endpoints.query;
+    const currentTable = allSchemas.find(
+      t => t.table_name === originalTable && t.table_schema === currentSchema
+    );
+    const columnTypeInfo = currentTable?.columns || [];

-    const reqBody = getDeleteQuery(pkClause, tableName, tableSchema, source);
+    if (!columnTypeInfo) {
+      throw new Error('Error in finding column info for table');
+    }
+
+    const sources = metadata.metadataObject?.sources;
+    const tableConfiguration = getTableConfiguration(tables, sources);
+
+    const {
+      endpoint,
+      getDeleteRowRequestBody,
+      processDeleteRowData,
+    } = dataSource.generateDeleteRowRequest();
+    const reqBody = getDeleteRowRequestBody({
+      pkClause,
+      tableName,
+      schemaName: tableSchema,
+      source,
+      columnInfo: columnTypeInfo,
+      tableConfiguration,
+    });

    const options = {
      method: 'POST',
@ -271,18 +279,27 @@ const deleteItem = (pkClause, tableName, tableSchema) => {
      headers: dataHeaders(getState),
      credentials: globalCookiePolicy,
    };
-    dispatch(requestAction(url, options)).then(
+    dispatch(requestAction(endpoint, options)).then(
      data => {
-        dispatch(vMakeTableRequests());
-        dispatch(
-          showSuccessNotification(
-            'Row deleted!',
-            'Affected rows: ' + data.affected_rows
-          )
-        );
+        try {
+          const affectedRows = processDeleteRowData(
+            data,
+            tableName,
+            tableSchema
+          );
+          dispatch(vMakeTableRequests());
+          dispatch(
+            showSuccessNotification(
+              'Row deleted!',
+              'Affected rows: ' + affectedRows
+            )
+          );
+        } catch (err) {
+          showError(err, 'Deleting row failed!', dispatch);
+        }
      },
      err => {
-        dispatch(showErrorNotification('Deleting row failed!', err.error, err));
+        showError(err, 'Deleting row failed!', dispatch);
      }
    );
  };
@ -296,30 +313,64 @@ const deleteItems = (pkClauses, tableName, tableSchema) => {
      return;
    }
    const source = getState().tables.currentDataSource;
+    const {
+      endpoint,
+      getBulkDeleteRowRequestBody,
+      processBulkDeleteRowData,
+    } = dataSource.generateBulkDeleteRowRequest();

-    const reqBody = {
-      type: 'bulk',
+    const { tables, metadata } = getState();
+    const { currentTable: originalTable, allSchemas, currentSchema } = tables;
+
+    const currentTable = allSchemas.find(
+      t => t.table_name === originalTable && t.table_schema === currentSchema
+    );
+    const columnTypeInfo = currentTable?.columns || [];
+
+    if (!columnTypeInfo) {
+      throw new Error('Error in finding column info for table');
+    }
+
+    const sources = metadata.metadataObject?.sources;
+    const tableConfiguration = getTableConfiguration(tables, sources);
+
+    const reqBody = getBulkDeleteRowRequestBody({
+      pkClauses,
+      tableName,
+      schemaName: tableSchema,
      source,
-      args: getBulkDeleteQuery(pkClauses, tableName, tableSchema, source),
-    };
+      columnInfo: columnTypeInfo,
+      tableConfiguration,
+    });
+
    const options = {
      method: 'POST',
      body: JSON.stringify(reqBody),
      headers: dataHeaders(getState),
      credentials: globalCookiePolicy,
    };
-    dispatch(requestAction(Endpoints.query, options)).then(
+
+    dispatch(requestAction(endpoint, options)).then(
      data => {
-        const affected = data.reduce((acc, d) => acc + d.affected_rows, 0);
-        dispatch(vMakeTableRequests());
-        dispatch(
-          showSuccessNotification('Rows deleted!', 'Affected rows: ' + affected)
-        );
+        try {
+          const affected = processBulkDeleteRowData(
+            data,
+            tableName,
+            tableSchema
+          );
+          dispatch(vMakeTableRequests());
+          dispatch(
+            showSuccessNotification(
+              'Rows deleted!',
+              'Affected rows: ' + affected
+            )
+          );
+        } catch (err) {
+          showError(err, 'Deleting rows failed!', dispatch);
+        }
      },
      err => {
-        dispatch(
-          showErrorNotification('Deleting rows failed!', err.error, err)
-        );
+        showError(err, 'Deleting rows failed!', dispatch);
      }
    );
  };
--- a/console/src/components/Services/Data/TableBrowseRows/utils.ts
+++ b/console/src/components/Services/Data/TableBrowseRows/utils.ts
@ -1,3 +1,6 @@
+import { MetadataDataSource } from '../../../../metadata/types';
+import { ReduxState } from './../../../../types';
+
 type TableSchema = {
  primary_key?: { columns: string[] };
  columns: Array<{ column_name: string }>;
@ -15,6 +18,22 @@ export const isTableWithPK = (
  );
 };

+export const getTableConfiguration = (
+  tables: ReduxState['tables'],
+  sources: MetadataDataSource[]
+) => {
+  const {
+    currentSchema,
+    currentTable: originalTable,
+    currentDataSource,
+  } = tables;
+  return sources
+    ?.find(s => s.name === currentDataSource)
+    ?.tables.find(
+      t => originalTable === t.table.name && currentSchema === t.table.schema
+    )?.configuration;
+};
+
 export const compareRows = (
  row1: Record<string, any>,
  row2: Record<string, any>,
--- a/console/src/components/Services/Data/TableInsertItem/InsertActions.js
+++ b/console/src/components/Services/Data/TableInsertItem/InsertActions.js
@ -7,16 +7,21 @@ import {
  showNotification,
 } from '../../Common/Notification';
 import dataHeaders from '../Common/Headers';
-import { getEnumColumnMappings, dataSource } from '../../../../dataSources';
+import {
+  getEnumColumnMappings,
+  dataSource,
+  findTable,
+} from '../../../../dataSources';
 import { getEnumOptionsQuery } from '../../../Common/utils/v1QueryUtils';
+import { isStringArray } from '../../../Common/utils/jsUtils';
 import {
  getInsertUpQuery,
  getInsertDownQuery,
 } from '../../../Common/utils/v1QueryUtils';
-import { isStringArray } from '../../../Common/utils/jsUtils';
 import { makeMigrationCall } from '../DataActions';
 import { removeAll } from 'react-notification-system-redux';
 import { getNotificationDetails } from '../../Common/Notification';
+import { getTableConfiguration } from '../TableBrowseRows/utils';

 const I_SET_CLONE = 'InsertItem/I_SET_CLONE';
 const I_RESET = 'InsertItem/I_RESET';
@ -87,26 +92,27 @@ const insertItem = (tableName, colValues, isMigration = false) => {
    /* Type all the values correctly */
    dispatch({ type: I_ONGOING_REQ });
    const insertObject = {};
-    const state = getState();
-    const { currentSchema, currentDataSource } = state.tables;
+    const { tables, metadata } = getState();
+    const { currentSchema, currentDataSource, allSchemas } = tables;
    const tableDef = { name: tableName, schema: currentSchema };
-
-    const currentTableInfo = state.tables.allSchemas.find(
-      t => t.table_name === tableName && t.table_schema === currentSchema
-    );
+    const sources = metadata.metadataObject?.sources;
+    const tableConfiguration = getTableConfiguration(tables, sources);
+    const currentTableInfo = findTable(allSchemas, tableDef);
+    if (!currentTableInfo) return;
    const columns = currentTableInfo.columns;
    let error = false;
    let errorMessage = '';
    Object.keys(colValues).map(colName => {
      const colSchema = columns.find(x => x.column_name === colName);
      const colType = colSchema.data_type;
+      const colValue = colValues[colName];
+
      if (Reals.indexOf(colType) > 0) {
-        insertObject[colName] =
-          parseFloat(colValues[colName], 10) || colValues[colName];
+        insertObject[colName] = parseFloat(colValue, 10) || colValue;
      } else if (colType === dataSource.columnDataTypes.BOOLEAN) {
-        if (colValues[colName] === 'true') {
+        if (colValue === 'true') {
          insertObject[colName] = true;
-        } else if (colValues[colName] === 'false') {
+        } else if (colValue === 'false') {
          insertObject[colName] = false;
        } else {
          insertObject[colName] = null;
@ -116,32 +122,24 @@ const insertItem = (tableName, colValues, isMigration = false) => {
        colType === dataSource.columnDataTypes.JSONB
      ) {
        try {
-          const val = JSON.parse(colValues[colName]);
+          const val = JSON.parse(colValue);
          insertObject[colName] = val;
        } catch (e) {
-          errorMessage =
-            colName +
-            ' :: could not read ' +
-            colValues[colName] +
-            ' as a valid JSON object/array';
+          errorMessage = `${colName} :: could not read ${colValue} as a valid JSON object/array`;
          error = true;
        }
      } else if (
        colType === dataSource.columnDataTypes.ARRAY &&
-        isStringArray(colValues[colName])
+        isStringArray(colValue)
      ) {
        try {
-          const arr = JSON.parse(colValues[colName]);
+          const arr = JSON.parse(colValue);
          insertObject[colName] = dataSource.arrayToPostgresArray(arr);
        } catch {
-          errorMessage =
-            colName +
-            ' :: could not read ' +
-            colValues[colName] +
-            ' as a valid array';
+          errorMessage = `${colName} :: could not read ${colValue} as a valid JSON object/array`;
        }
      } else {
-        insertObject[colName] = colValues[colName];
+        insertObject[colName] = colValue;
      }
    });

@ -153,22 +151,27 @@ const insertItem = (tableName, colValues, isMigration = false) => {
      });
    }
    const returning = columns.map(col => col.column_name);
-    const reqBody = {
-      type: 'insert',
-      args: {
-        source: currentDataSource,
-        table: tableDef,
-        objects: [insertObject],
-        returning,
-      },
-    };
+    if (!dataSource.generateInsertRequest) return;
+    const {
+      getInsertRequestBody,
+      processInsertData,
+      endpoint: url,
+    } = dataSource.generateInsertRequest();
+
+    const reqBody = getInsertRequestBody({
+      source: currentDataSource,
+      tableDef,
+      tableConfiguration,
+      insertObject,
+      returning,
+    });
+
    const options = {
      method: 'POST',
      credentials: globalCookiePolicy,
      headers: dataHeaders(getState),
      body: JSON.stringify(reqBody),
    };
-    const url = Endpoints.query;

    const migrationSuccessCB = (affectedRows, returnedFields) => {
      const detailsAction = {
@ -205,19 +208,26 @@ const insertItem = (tableName, colValues, isMigration = false) => {
      requestAction(url, options, I_REQUEST_SUCCESS, I_REQUEST_ERROR)
    ).then(
      data => {
-        const affectedRows = data.affected_rows;
+        const { affectedRows, returnedFields } = processInsertData(
+          data,
+          tableConfiguration,
+          {
+            currentSchema,
+            currentTable: tableName,
+          }
+        );
        if (isMigration) {
          dispatch(
            insertItemAsMigration(
              tableDef,
-              data.returning[0],
+              returnedFields,
              currentTableInfo.primary_key,
              columns,
-              () => migrationSuccessCB(affectedRows, data.returning[0])
+              () => migrationSuccessCB(affectedRows, returnedFields)
            )
          );
        } else {
-          migrationSuccessCB(affectedRows, data.returning[0]);
+          migrationSuccessCB(affectedRows, returnedFields);
        }
      },

--- a/console/src/components/Services/Data/TableModify/ModifyView.js
+++ b/console/src/components/Services/Data/TableModify/ModifyView.js
@ -267,10 +267,21 @@ const ModifyView = props => {
            <hr />
            {getViewColumnsSection()}
            <hr />
-            <ComputedFields tableSchema={tableSchema} />
-            <hr />
-            <RootFields tableSchema={tableSchema} />
-            <hr />
+
+            {isFeatureSupported('tables.modify.computedFields') && (
+              <>
+                <ComputedFields tableSchema={tableSchema} />
+                <hr />
+              </>
+            )}
+
+            {isFeatureSupported('tables.modify.customGqlRoot') && (
+              <>
+                <RootFields tableSchema={tableSchema} />
+                <hr />
+              </>
+            )}
+
            {untrackBtn}
            {deleteBtn}
            <br />
--- a/console/src/components/Services/Data/TableRelationships/autoRelations.js
+++ b/console/src/components/Services/Data/TableRelationships/autoRelations.js
@ -1,3 +1,5 @@
+import { isRelationshipValid as isCitusRelValid } from '../../../../dataSources/services/citus/utils';
+
 const sameRelCols = (currCols, existingCols) => {
  return currCols.sort().join(',') === existingCols.sort().join(',');
 };
@ -71,6 +73,22 @@ const isExistingArrRel = (currentArrRels, relCols, relTable) => {
  return _isExistingArrRel;
 };

+const isRelationshipValid = (rel, allSchemas) => {
+  const lTable = allSchemas.find(
+    t => t.table_name === rel.lTable && t.table_schema === rel.lSchema
+  );
+  const rTable = allSchemas.find(
+    t => t.table_name === rel.rTable && t.table_schema === rel.rSchema
+  );
+
+  /* valid relationship rules for citus */
+  if (lTable?.citus_table_type && rTable?.citus_table_type) {
+    return isCitusRelValid(rel, lTable, rTable);
+  }
+
+  return true;
+};
+
 const suggestedRelationshipsRaw = (tableName, allSchemas, currentSchema) => {
  const objRels = [];
  const arrRels = [];
@ -177,10 +195,10 @@ const suggestedRelationshipsRaw = (tableName, allSchemas, currentSchema) => {
  for (let i = 0; i < length; i++) {
    const objRel = objRels[i] ? objRels[i] : null;
    const arrRel = arrRels[i] ? arrRels[i] : null;
-    if (objRel !== null) {
+    if (objRel !== null && isRelationshipValid(objRel, allSchemas)) {
      finalObjRel.push(objRel);
    }
-    if (arrRel !== null) {
+    if (arrRel !== null && isRelationshipValid(arrRel, allSchemas)) {
      finalArrayRel.push(arrRel);
    }
  }
--- a/console/src/components/Services/Data/tests/snapshots/mergeData.test.ts.snap
+++ b/console/src/components/Services/Data/tests/snapshots/mergeData.test.ts.snap
@ -0,0 +1,236 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`verify merge data for citus citus with no tables 1`] = `Array []`;
+
+exports[`verify merge data for citus citus with relationships and foreign key 1`] = `
+Array [
+  Object {
+    "check_constraints": Array [],
+    "citus_table_type": "local",
+    "columns": Array [
+      Object {
+        "column_default": "nextval('users_id_seq'::regclass)",
+        "column_name": "id",
+        "comment": null,
+        "data_type": "integer",
+        "data_type_name": "int4",
+        "is_nullable": "NO",
+        "ordinal_position": 1,
+        "table_name": "users",
+        "table_schema": "public",
+      },
+      Object {
+        "column_default": null,
+        "column_name": "name",
+        "comment": null,
+        "data_type": "text",
+        "data_type_name": "text",
+        "is_nullable": "NO",
+        "ordinal_position": 2,
+        "table_name": "users",
+        "table_schema": "public",
+      },
+    ],
+    "comment": null,
+    "computed_fields": Array [],
+    "configuration": Object {},
+    "foreign_key_constraints": Array [],
+    "is_enum": false,
+    "is_table_tracked": true,
+    "opp_foreign_key_constraints": Array [
+      Object {
+        "column_mapping": Object {
+          "user_id": "id",
+        },
+        "constraint_name": "posts_user_id_fkey",
+        "is_ref_table_tracked": true,
+        "is_table_tracked": true,
+        "on_delete": "r",
+        "on_update": "r",
+        "ref_table": "users",
+        "ref_table_table_schema": "public",
+        "table_name": "posts",
+        "table_schema": "public",
+      },
+    ],
+    "permissions": Array [],
+    "primary_key": Object {
+      "columns": Array [
+        "id",
+      ],
+      "constraint_name": "users_pkey",
+      "table_name": "users",
+      "table_schema": "public",
+    },
+    "relationships": Array [
+      Object {
+        "rel_def": Object {
+          "foreign_key_constraint_on": Object {
+            "column": "user_id",
+            "table": Object {
+              "name": "posts",
+              "schema": "public",
+            },
+          },
+        },
+        "rel_name": "posts",
+        "rel_type": "array",
+        "table_name": "users",
+        "table_schema": "public",
+      },
+    ],
+    "remote_relationships": Array [],
+    "table_name": "users",
+    "table_schema": "public",
+    "table_type": "TABLE",
+    "triggers": Array [],
+    "unique_constraints": Array [],
+    "view_info": null,
+  },
+  Object {
+    "check_constraints": Array [],
+    "citus_table_type": "local",
+    "columns": Array [
+      Object {
+        "column_default": "nextval('posts_id_seq'::regclass)",
+        "column_name": "id",
+        "comment": null,
+        "data_type": "integer",
+        "data_type_name": "int4",
+        "is_nullable": "NO",
+        "ordinal_position": 1,
+        "table_name": "posts",
+        "table_schema": "public",
+      },
+      Object {
+        "column_default": null,
+        "column_name": "user_id",
+        "comment": null,
+        "data_type": "integer",
+        "data_type_name": "int4",
+        "is_nullable": "NO",
+        "ordinal_position": 3,
+        "table_name": "posts",
+        "table_schema": "public",
+      },
+      Object {
+        "column_default": null,
+        "column_name": "post",
+        "comment": null,
+        "data_type": "text",
+        "data_type_name": "text",
+        "is_nullable": "NO",
+        "ordinal_position": 2,
+        "table_name": "posts",
+        "table_schema": "public",
+      },
+    ],
+    "comment": null,
+    "computed_fields": Array [],
+    "configuration": Object {},
+    "foreign_key_constraints": Array [
+      Object {
+        "column_mapping": Object {
+          "user_id": "id",
+        },
+        "constraint_name": "posts_user_id_fkey",
+        "is_ref_table_tracked": true,
+        "is_table_tracked": true,
+        "on_delete": "r",
+        "on_update": "r",
+        "ref_table": "users",
+        "ref_table_table_schema": "public",
+        "table_name": "posts",
+        "table_schema": "public",
+      },
+    ],
+    "is_enum": false,
+    "is_table_tracked": true,
+    "opp_foreign_key_constraints": Array [],
+    "permissions": Array [],
+    "primary_key": Object {
+      "columns": Array [
+        "id",
+      ],
+      "constraint_name": "posts_pkey",
+      "table_name": "posts",
+      "table_schema": "public",
+    },
+    "relationships": Array [
+      Object {
+        "rel_def": Object {
+          "foreign_key_constraint_on": "user_id",
+        },
+        "rel_name": "user",
+        "rel_type": "object",
+        "table_name": "posts",
+        "table_schema": "public",
+      },
+    ],
+    "remote_relationships": Array [],
+    "table_name": "posts",
+    "table_schema": "public",
+    "table_type": "TABLE",
+    "triggers": Array [],
+    "unique_constraints": Array [],
+    "view_info": null,
+  },
+]
+`;
+
+exports[`verify merge data for citus citus with table 1`] = `
+Array [
+  Object {
+    "check_constraints": Array [],
+    "citus_table_type": "local",
+    "columns": Array [
+      Object {
+        "column_default": "nextval('users_id_seq'::regclass)",
+        "column_name": "id",
+        "comment": null,
+        "data_type": "integer",
+        "data_type_name": "int4",
+        "is_nullable": "NO",
+        "ordinal_position": 1,
+        "table_name": "users",
+        "table_schema": "public",
+      },
+      Object {
+        "column_default": null,
+        "column_name": "name",
+        "comment": null,
+        "data_type": "text",
+        "data_type_name": "text",
+        "is_nullable": "NO",
+        "ordinal_position": 2,
+        "table_name": "users",
+        "table_schema": "public",
+      },
+    ],
+    "comment": null,
+    "computed_fields": Array [],
+    "configuration": Object {},
+    "foreign_key_constraints": Array [],
+    "is_enum": false,
+    "is_table_tracked": true,
+    "opp_foreign_key_constraints": Array [],
+    "permissions": Array [],
+    "primary_key": Object {
+      "columns": Array [
+        "id",
+      ],
+      "constraint_name": "users_pkey",
+      "table_name": "users",
+      "table_schema": "public",
+    },
+    "relationships": Array [],
+    "remote_relationships": Array [],
+    "table_name": "users",
+    "table_schema": "public",
+    "table_type": "TABLE",
+    "triggers": Array [],
+    "unique_constraints": Array [],
+    "view_info": null,
+  },
+]
+`;
--- a/console/src/components/Services/Data/tests/fixtures/input.ts
+++ b/console/src/components/Services/Data/tests/fixtures/input.ts
@ -0,0 +1,93 @@
+export const citus_no_tables = {
+  query_data: [
+    { result_type: 'TuplesOk', result: [['tables'], ['[]']] },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+  ] as any,
+  metadata: [],
+};
+
+export const citus_with_table = {
+  query_data: [
+    {
+      result_type: 'TuplesOk',
+      result: [
+        ['tables'],
+        [
+          '[{"table_schema":"public","table_name":"users","table_type":"TABLE","comment":null,"columns":[{"comment": null, "data_type": "integer", "table_name": "users", "column_name": "id", "is_nullable": "NO", "table_schema": "public", "column_default": "nextval(\'users_id_seq\'::regclass)", "data_type_name": "int4", "ordinal_position": 1}, {"comment": null, "data_type": "text", "table_name": "users", "column_name": "name", "is_nullable": "NO", "table_schema": "public", "column_default": null, "data_type_name": "text", "ordinal_position": 2}],"triggers":[],"view_info":null,"citus_table_type":"local"}]',
+        ],
+      ],
+    },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+    {
+      result_type: 'TuplesOk',
+      result: [
+        ['coalesce'],
+        [
+          '[{"table_schema":"public","table_name":"users","constraint_name":"users_pkey","columns":["id"]}]',
+        ],
+      ],
+    },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+  ] as any,
+  metadata: [{ table: { schema: 'public', name: 'users' } }],
+};
+
+export const citus_with_relationships_fk = {
+  query_data: [
+    {
+      result_type: 'TuplesOk',
+      result: [
+        ['tables'],
+        [
+          '[{"table_schema":"public","table_name":"users","table_type":"TABLE","comment":null,"columns":[{"comment": null, "data_type": "integer", "table_name": "users", "column_name": "id", "is_nullable": "NO", "table_schema": "public", "column_default": "nextval(\'users_id_seq\'::regclass)", "data_type_name": "int4", "ordinal_position": 1}, {"comment": null, "data_type": "text", "table_name": "users", "column_name": "name", "is_nullable": "NO", "table_schema": "public", "column_default": null, "data_type_name": "text", "ordinal_position": 2}],"triggers":[],"view_info":null,"citus_table_type":"local"}, {"table_schema":"public","table_name":"posts","table_type":"TABLE","comment":null,"columns":[{"comment": null, "data_type": "integer", "table_name": "posts", "column_name": "id", "is_nullable": "NO", "table_schema": "public", "column_default": "nextval(\'posts_id_seq\'::regclass)", "data_type_name": "int4", "ordinal_position": 1}, {"comment": null, "data_type": "integer", "table_name": "posts", "column_name": "user_id", "is_nullable": "NO", "table_schema": "public", "column_default": null, "data_type_name": "int4", "ordinal_position": 3}, {"comment": null, "data_type": "text", "table_name": "posts", "column_name": "post", "is_nullable": "NO", "table_schema": "public", "column_default": null, "data_type_name": "text", "ordinal_position": 2}],"triggers":[],"view_info":null,"citus_table_type":"local"}]',
+        ],
+      ],
+    },
+    {
+      result_type: 'TuplesOk',
+      result: [
+        ['coalesce'],
+        [
+          '[{"table_schema":"public","table_name":"posts","constraint_name":"posts_user_id_fkey","ref_table_table_schema":"public","ref_table":"users","column_mapping":{ "user_id" : "id" },"on_update":"r","on_delete":"r"}]',
+        ],
+      ],
+    },
+    {
+      result_type: 'TuplesOk',
+      result: [
+        ['coalesce'],
+        [
+          '[{"table_schema":"public","table_name":"posts","constraint_name":"posts_pkey","columns":["id"]}, {"table_schema":"public","table_name":"users","constraint_name":"users_pkey","columns":["id"]}]',
+        ],
+      ],
+    },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+    { result_type: 'TuplesOk', result: [['coalesce'], ['[]']] },
+  ] as any,
+  metadata: [
+    {
+      table: { schema: 'public', name: 'posts' },
+      object_relationships: [
+        { name: 'user', using: { foreign_key_constraint_on: 'user_id' } },
+      ],
+    },
+    {
+      table: { schema: 'public', name: 'users' },
+      array_relationships: [
+        {
+          name: 'posts',
+          using: {
+            foreign_key_constraint_on: {
+              column: 'user_id',
+              table: { schema: 'public', name: 'posts' },
+            },
+          },
+        },
+      ],
+    },
+  ],
+};
--- a/console/src/components/Services/Data/tests/mergeData.test.ts
+++ b/console/src/components/Services/Data/tests/mergeData.test.ts
@ -0,0 +1,26 @@
+import { mergeDataCitus } from '../mergeData';
+import {
+  citus_no_tables,
+  citus_with_table,
+  citus_with_relationships_fk,
+} from './fixtures/input';
+
+describe('verify merge data for citus', () => {
+  test('citus with no tables', () => {
+    const { query_data, metadata } = citus_no_tables;
+    const res = mergeDataCitus(query_data, metadata);
+    expect(res).toMatchSnapshot();
+  });
+
+  test('citus with table', () => {
+    const { query_data, metadata } = citus_with_table;
+    const res = mergeDataCitus(query_data, metadata);
+    expect(res).toMatchSnapshot();
+  });
+
+  test('citus with relationships and foreign key', () => {
+    const { query_data, metadata } = citus_with_relationships_fk;
+    const res = mergeDataCitus(query_data, metadata);
+    expect(res).toMatchSnapshot();
+  });
+});
--- a/console/src/components/Services/Data/mergeData.ts
+++ b/console/src/components/Services/Data/mergeData.ts
@ -2,6 +2,7 @@
 import { Table } from '../../../dataSources/types';
 import { TableEntry } from '../../../metadata/types';
 import { PostgresTable } from '../../../dataSources/services/postgresql/types';
+import { CitusTable } from '../../../dataSources/services/citus/types';
 import { dataSource } from '../../../dataSources';
 import { FixMe } from '../../../types';

@ -647,3 +648,184 @@ export const mergeDataBigQuery = (
  });
  return result;
 };
+
+export const mergeDataCitus = (
+  data: Array<{ result: string[] }>,
+  metadataTables: TableEntry[]
+): Table[] => {
+  const tableList = JSON.parse(data[0].result[1]) as CitusTable[];
+  const fkList = JSON.parse(data[1].result[1]) as Omit<
+    Table['foreign_key_constraints'][0],
+    'is_table_tracked' | 'is_ref_table_tracked'
+  >[];
+  const primaryKeys = JSON.parse(data[2].result[1]) as Table['primary_key'][];
+  const uniqueKeys = JSON.parse(data[3].result[1]) as {
+    table_name: string;
+    table_schema: string;
+    constraint_name: string;
+    columns: string[];
+  }[];
+  const checkConstraints = dataSource?.checkConstraintsSql
+    ? (JSON.parse(data[4].result[1]) as Table['check_constraints'])
+    : ([] as Table['check_constraints']);
+  const _mergedTableData: Table[] = [];
+
+  const trackedFkData = fkList.map(fk => ({
+    ...fk,
+    is_table_tracked: !!metadataTables.some(
+      t => t.table.name === fk.table_name && t.table.schema === fk.table_schema
+    ),
+    is_ref_table_tracked: !!metadataTables.some(
+      t =>
+        t.table.name === fk.ref_table &&
+        t.table.schema === fk.ref_table_table_schema
+    ),
+  }));
+
+  tableList.forEach(infoSchemaTableInfo => {
+    const tableSchema = infoSchemaTableInfo.table_schema;
+    const tableName = infoSchemaTableInfo.table_name;
+    const metadataTable = metadataTables?.find(
+      t => t.table.schema === tableSchema && t.table.name === tableName
+    );
+
+    const columns = infoSchemaTableInfo.columns;
+    const comment = infoSchemaTableInfo.comment;
+    const tableType = infoSchemaTableInfo.table_type;
+    const triggers = infoSchemaTableInfo.triggers;
+    const viewInfo = infoSchemaTableInfo.view_info;
+    const citus_table_type = infoSchemaTableInfo.citus_table_type;
+
+    const keys =
+      primaryKeys.find(
+        key => key?.table_name === tableName && key.table_schema === tableSchema
+      ) || null;
+
+    const unique =
+      uniqueKeys.filter(
+        (key: any) =>
+          key?.table_name === tableName && key.table_schema === tableSchema
+      ) || [];
+
+    const check =
+      checkConstraints.filter(
+        (key: any) =>
+          key?.table_name === tableName && key.table_schema === tableSchema
+      ) || [];
+
+    const permissions: Table['permissions'] = [];
+    let fkConstraints: Table['foreign_key_constraints'] = [];
+    let refFkConstraints: Table['foreign_key_constraints'] = [];
+    let remoteRelationships: Table['remote_relationships'] = [];
+    let isEnum = false;
+    let configuration = {};
+    let computed_fields: Table['computed_fields'] = [];
+    const relationships: Table['relationships'] = [];
+
+    if (metadataTable) {
+      isEnum = metadataTable?.is_enum ?? false;
+      configuration = metadataTable?.configuration ?? {};
+
+      fkConstraints = trackedFkData.filter(
+        (fk: any) =>
+          fk.table_schema === tableSchema && fk.table_name === tableName
+      );
+
+      refFkConstraints = trackedFkData.filter(
+        (fk: any) =>
+          fk.ref_table_table_schema === tableSchema &&
+          fk.ref_table === tableName &&
+          fk.is_ref_table_tracked
+      );
+
+      remoteRelationships = (metadataTable?.remote_relationships ?? []).map(
+        ({ definition, name }) => ({
+          remote_relationship_name: name,
+          table_name: tableName,
+          table_schema: tableSchema,
+          definition,
+        })
+      );
+
+      computed_fields = (metadataTable?.computed_fields ?? []).map(field => ({
+        comment: field.comment || '',
+        computed_field_name: field.name,
+        name: field.name,
+        table_name: tableName,
+        table_schema: tableSchema,
+        definition: field.definition as Table['computed_fields'][0]['definition'],
+      }));
+
+      metadataTable?.array_relationships?.forEach(rel => {
+        relationships.push({
+          rel_def: rel.using,
+          rel_name: rel.name,
+          table_name: tableName,
+          table_schema: tableSchema,
+          rel_type: 'array',
+        });
+      });
+
+      metadataTable?.object_relationships?.forEach(rel => {
+        relationships.push({
+          rel_def: rel.using,
+          rel_name: rel.name,
+          table_name: tableName,
+          table_schema: tableSchema,
+          rel_type: 'object',
+        });
+      });
+
+      const rolePermMap: Record<string, any> = {};
+
+      permKeys.forEach(key => {
+        if (metadataTable) {
+          metadataTable[key]?.forEach((perm: any) => {
+            rolePermMap[perm.role] = {
+              permissions: {
+                ...(rolePermMap[perm.role] &&
+                  rolePermMap[perm.role].permissions),
+                [keyToPermission[key]]: perm.permission,
+              },
+            };
+          });
+        }
+      });
+
+      Object.keys(rolePermMap).forEach(role => {
+        permissions.push({
+          role_name: role,
+          permissions: rolePermMap[role].permissions,
+          table_name: tableName,
+          table_schema: tableSchema,
+        });
+      });
+    }
+
+    const _mergedInfo = {
+      table_schema: tableSchema,
+      table_name: tableName,
+      table_type: tableType as Table['table_type'],
+      is_table_tracked: !!metadataTable,
+      columns,
+      comment,
+      triggers,
+      primary_key: keys,
+      relationships,
+      permissions,
+      unique_constraints: unique,
+      check_constraints: check,
+      foreign_key_constraints: fkConstraints,
+      opp_foreign_key_constraints: refFkConstraints,
+      view_info: viewInfo as Table['view_info'],
+      remote_relationships: remoteRelationships,
+      is_enum: isEnum,
+      configuration: configuration as Table['configuration'],
+      computed_fields,
+      citus_table_type,
+    };
+
+    _mergedTableData.push(_mergedInfo);
+  });
+  return _mergedTableData;
+};
--- a/console/src/dataSources/common/graphqlUtil.ts
+++ b/console/src/dataSources/common/graphqlUtil.ts
@ -0,0 +1,186 @@
+import { CustomRootFields, TableConfig } from './../../metadata/types';
+import {
+  OrderBy,
+  WhereClause,
+} from '../../components/Common/utils/v1QueryUtils';
+import { ReduxState } from '../../types';
+import { BaseTableColumn, Relationship, Table } from '../types';
+
+type Tables = ReduxState['tables'];
+
+export interface QueryBody {
+  clauses: string;
+  relationshipInfo: Relationship[];
+}
+export interface GetGraphQLQuery {
+  allSchemas: Table[];
+  view: Tables['view'];
+  originalTable: string;
+  currentSchema: string;
+  isExport?: boolean;
+  tableConfiguration: TableConfig;
+  queryBody: (config: QueryBody) => string;
+  getFormattedValue: (type: string, value: any) => string | number | undefined;
+}
+interface GetFullQueryName {
+  tableName: string;
+  schema: string;
+  tableConfiguration: TableConfig;
+  defaultSchema?: string;
+  operation: keyof Omit<
+    CustomRootFields,
+    'select_by_pk' | 'insert_one' | 'update_by_pk' | 'delete_by_pk'
+  >;
+}
+
+const generateSortClauseQueryString = (
+  sorts: OrderBy[],
+  tableConfiguration: TableConfig
+): string | null => {
+  const customColumns = tableConfiguration?.custom_column_names ?? {};
+  const sortClausesArr = sorts.map((i: OrderBy) => {
+    return `${customColumns[i.column] ?? i.column}: ${i.type}`;
+  });
+  return sortClausesArr.length
+    ? `order_by: {${sortClausesArr.join(',')}}`
+    : null;
+};
+
+export const getColQuery = (
+  cols: (string | { name: string; columns: string[] })[],
+  limit: number,
+  relationships: Relationship[],
+  tableConfiguration: TableConfig
+): string[] => {
+  return cols.map(c => {
+    const customColumns = tableConfiguration?.custom_column_names ?? {};
+    if (typeof c === 'string') return customColumns[c] ?? c;
+    const rel = relationships.find((r: any) => r.rel_name === c.name);
+    return `${customColumns[c.name] ?? c.name} ${
+      rel?.rel_type === 'array' ? `(limit: ${limit})` : ''
+    } { 
+        ${getColQuery(c.columns, limit, relationships, tableConfiguration).join(
+          '\n'
+        )} }`;
+  });
+};
+
+export const operators = [
+  { name: 'equals', value: '$eq', graphqlOp: '_eq' },
+  { name: 'not equals', value: '$ne', graphqlOp: '_neq' },
+  { name: '>', value: '$gt', graphqlOp: '_gt' },
+  { name: '<', value: '$lt', graphqlOp: '_lt' },
+  { name: '>=', value: '$gte', graphqlOp: '_gte' },
+  { name: '<=', value: '$lte', graphqlOp: '_lte' },
+];
+
+export const RqlToGraphQlOp = (op: string) => {
+  if (!op || !op?.startsWith('$')) return 'none';
+  return (
+    operators.find(_op => _op.value === op)?.graphqlOp ?? op.replace('$', '_')
+  );
+};
+
+export const generateWhereClauseQueryString = (
+  wheres: WhereClause[],
+  columnTypeInfo: BaseTableColumn[],
+  tableConfiguration: TableConfig,
+  getFormattedValue: (type: string, value: any) => string | number | undefined
+): string | null => {
+  const customColumns = tableConfiguration?.custom_column_names ?? {};
+  const whereClausesArr = wheres.map((i: Record<string, any>) => {
+    const columnName = Object.keys(i)[0];
+    const RqlOperator = Object.keys(i[columnName])[0];
+    const value = i[columnName][RqlOperator];
+    const type = columnTypeInfo?.find(c => c.column_name === columnName)
+      ?.data_type_name;
+    return `${customColumns[columnName] ?? columnName}: {${RqlToGraphQlOp(
+      RqlOperator
+    )}: ${getFormattedValue(type || 'varchar', value)} }`;
+  });
+  return whereClausesArr.length
+    ? `where: {${whereClausesArr.join(',')}}`
+    : null;
+};
+
+export const getGraphQLQueryBase = ({
+  allSchemas,
+  view,
+  originalTable,
+  currentSchema,
+  isExport = false,
+  tableConfiguration,
+  queryBody,
+  getFormattedValue,
+}: GetGraphQLQuery) => {
+  const currentTable: Table | undefined = allSchemas?.find(
+    (t: Table) =>
+      t.table_name === originalTable && t.table_schema === currentSchema
+  );
+  const columnTypeInfo: BaseTableColumn[] = currentTable?.columns || [];
+  const relationshipInfo: Relationship[] = currentTable?.relationships || [];
+
+  if (!columnTypeInfo) {
+    throw new Error('Error in finding column info for table');
+  }
+
+  let whereConditions: WhereClause[] = [];
+  let isRelationshipView = false;
+  if (view.query?.where) {
+    if (view.query.where.$and) {
+      whereConditions = view.query.where.$and;
+    } else {
+      isRelationshipView = true;
+      whereConditions = Object.keys(view.query.where)
+        .filter(k => view.query.where[k])
+        .map(k => {
+          const obj = {} as any;
+          obj[k] = { $eq: view.query.where[k] };
+          return obj;
+        });
+    }
+  }
+  const sortConditions: OrderBy[] = [];
+  if (view.query?.order_by) {
+    sortConditions.push(...view.query.order_by);
+  }
+  const limit =
+    isExport || !view.curFilter?.limit
+      ? null
+      : `limit: ${view.curFilter.limit}`;
+  const offset = isExport
+    ? null
+    : `offset: ${!isRelationshipView ? view.curFilter?.offset ?? 0 : 0}`;
+  const clauses = `${[
+    generateWhereClauseQueryString(
+      whereConditions,
+      columnTypeInfo,
+      tableConfiguration,
+      getFormattedValue
+    ),
+    generateSortClauseQueryString(sortConditions, tableConfiguration),
+    limit,
+    offset,
+  ]
+    .filter(Boolean)
+    .join(',')}`;
+
+  return queryBody({ clauses, relationshipInfo });
+};
+
+export const getFullQueryNameBase = (defaultSchema: string) => ({
+  tableName,
+  schema,
+  tableConfiguration,
+  operation,
+}: GetFullQueryName): string => {
+  const customRootFields = tableConfiguration?.custom_root_fields ?? {};
+  if (customRootFields[operation]) return customRootFields[operation]!;
+  const withUpdate = operation === 'update' ? 'update_' : '';
+  const withSchema = schema === defaultSchema ? '' : `${schema}_`;
+  const withAgg = operation === 'select_aggregate' ? `_aggregate` : '';
+  const withDelete = operation === 'delete' ? 'delete_' : '';
+  const withInsert = operation === 'insert' ? 'insert_' : '';
+  const trackedTableName = tableConfiguration?.custom_name || tableName;
+  return `${withDelete}${withUpdate}${withInsert}${withSchema}${trackedTableName}${withAgg}`;
+};
--- a/console/src/dataSources/common/index.ts
+++ b/console/src/dataSources/common/index.ts
@ -1,8 +1,8 @@
-import { isEqual } from '../components/Common/utils/jsUtils';
-import { Nullable } from '../components/Common/utils/tsUtils';
-import { QualifiedTable } from '../metadata/types';
-import { FixMe } from '../types';
-import { BaseTable, CheckConstraint, Relationship, Table } from './types';
+import { isEqual } from '../../components/Common/utils/jsUtils';
+import { Nullable } from '../../components/Common/utils/tsUtils';
+import { QualifiedTable } from '../../metadata/types';
+import { FixMe } from '../../types';
+import { BaseTable, CheckConstraint, Relationship, Table } from '../types';

 export type Operations = 'insert' | 'select' | 'update' | 'delete';
 export const QUERY_TYPES: Operations[] = [
@ -434,3 +434,5 @@ export const getCheckConstraintBoolExp = (check: string) => {

  return check;
 };
+
+export * from './graphqlUtil';
--- a/console/src/dataSources/index.ts
+++ b/console/src/dataSources/index.ts
@ -13,6 +13,11 @@ import {
  SupportedFeaturesType,
  generateTableRowRequestType,
  BaseTableColumn,
+  generateInsertRequestType,
+  GenerateRowsCountRequestType,
+  GenerateEditRowRequest,
+  GenerateDeleteRowRequest,
+  GenerateBulkDeleteRowRequest,
  ViolationActions,
 } from './types';
 import { PGFunction, FunctionState } from './services/postgresql/types';
@ -22,8 +27,15 @@ import { QualifiedTable } from '../metadata/types';
 import { supportedFeatures as PGSupportedFeatures } from './services/postgresql';
 import { supportedFeatures as MssqlSupportedFeatures } from './services/mssql';
 import { supportedFeatures as BigQuerySupportedFeatures } from './services/bigquery';
+import { supportedFeatures as CitusQuerySupportedFeatures } from './services/citus';

-export const drivers = ['postgres', 'mysql', 'mssql', 'bigquery'] as const;
+export const drivers = [
+  'postgres',
+  'mysql',
+  'mssql',
+  'bigquery',
+  'citus',
+] as const;
 export type Driver = typeof drivers[number];

 export const driverToLabel: Record<Driver, string> = {
@ -31,6 +43,14 @@ export const driverToLabel: Record<Driver, string> = {
  postgres: 'PostgreSQL',
  mssql: 'MS SQL Server',
  bigquery: 'BigQuery',
+  citus: 'Citus',
+};
+
+export const sourceNames = {
+  postgres: PGSupportedFeatures?.driver?.name,
+  mssql: MssqlSupportedFeatures?.driver?.name,
+  bigquery: BigQuerySupportedFeatures?.driver?.name,
+  citus: CitusQuerySupportedFeatures?.driver?.name,
 };

 export type ColumnsInfoResult = {
@ -336,7 +356,12 @@ export interface DataSourcesAPI {
  supportedFeatures?: SupportedFeaturesType;
  violationActions: ViolationActions[];
  defaultRedirectSchema?: string;
+  generateInsertRequest?: () => generateInsertRequestType;
+  generateRowsCountRequest?: () => GenerateRowsCountRequestType;
  getPartitionDetailsSql?: (tableName: string, tableSchema: string) => string;
+  generateEditRowRequest?: () => GenerateEditRowRequest;
+  generateDeleteRowRequest?: () => GenerateDeleteRowRequest;
+  generateBulkDeleteRowRequest?: () => GenerateBulkDeleteRowRequest;
 }

 export let currentDriver: Driver = 'postgres';
@ -358,6 +383,7 @@ export const getSupportedDrivers = (
    PGSupportedFeatures,
    MssqlSupportedFeatures,
    BigQuerySupportedFeatures,
+    CitusQuerySupportedFeatures,
  ]
    .filter(d => isEnabled(d))
    .map(d => d.driver.name) as Driver[];
--- a/console/src/dataSources/services/bigquery/index.tsx
+++ b/console/src/dataSources/services/bigquery/index.tsx
@ -63,8 +63,11 @@ const operators = [
  { name: '<=', value: '$lte', graphqlOp: '_lte' },
 ];

+// createSQLRegex matches one or more sql for creating view, table or functions, and extracts the type, schema, name and also if it is a partition.
+// An example string it matches: CREATE TABLE myschema.user(id serial primary key, name text);
+// type = table, schema = myschema, nameWithSchema = user, partition = undefined
 // eslint-disable-next-line no-useless-escape
-const createSQLRegex = /create\s*(?:|or\s*replace)\s*(?<type>view|table|function)\s*(?:\s*if*\s*not\s*exists\s*)?((?<schema>\"?\w+\"?)\.(?<tableWithSchema>\"?\w+\"?)|(?<table>\"?\w+\"?))\s*(?<partition>partition\s*of)?/gim;
+const createSQLRegex = /create\s*(?:|or\s*replace)\s*(?<type>view|table|function)\s*(?:\s*if*\s*not\s*exists\s*)?((?<schema>\"?\w+\"?)\.(?<nameWithSchema>\"?\w+\"?)|(?<name>\"?\w+\"?))\s*(?<partition>partition\s*of)?/gim;

 export const displayTableName = (table: Table) => {
  const tableName = table.table_name;
--- a/console/src/dataSources/services/citus/index.tsx
+++ b/console/src/dataSources/services/citus/index.tsx
@ -0,0 +1,72 @@
+import { DataSourcesAPI } from '../..';
+import { getFetchTablesListQuery, schemaListSql } from './sqlUtils';
+import {
+  generateTableRowRequest,
+  generateInsertRequest,
+  generateRowsCountRequest,
+  generateEditRowRequest,
+  generateDeleteRowRequest,
+  generateBulkDeleteRowRequest,
+} from './utils';
+import { SupportedFeaturesType } from '../../types';
+import {
+  postgres,
+  supportedFeatures as PgSupportedFeatures,
+} from '../postgresql';
+
+export const supportedFeatures: SupportedFeaturesType = {
+  ...PgSupportedFeatures,
+  driver: {
+    name: 'citus',
+  },
+  tables: {
+    ...(PgSupportedFeatures?.tables ?? {}),
+    browse: {
+      enabled: true,
+      aggregation: true,
+      customPagination: true,
+    },
+    modify: {
+      ...(PgSupportedFeatures?.tables?.modify ?? {}),
+      enabled: true,
+      computedFields: true,
+      triggers: true,
+      customGqlRoot: false,
+      setAsEnum: false,
+      untrack: true,
+      delete: true,
+    },
+  },
+  events: {
+    triggers: {
+      enabled: true,
+      add: false,
+    },
+  },
+  actions: {
+    enabled: true,
+    relationships: false,
+  },
+  functions: {
+    enabled: true,
+    track: {
+      enabled: true,
+    },
+    nonTrackableFunctions: {
+      enabled: true,
+    },
+  },
+};
+
+export const citus: DataSourcesAPI = {
+  ...postgres,
+  supportedFeatures,
+  schemaListSql,
+  getFetchTablesListQuery,
+  generateTableRowRequest,
+  generateInsertRequest,
+  generateRowsCountRequest,
+  generateEditRowRequest,
+  generateDeleteRowRequest,
+  generateBulkDeleteRowRequest,
+};
--- a/console/src/dataSources/services/citus/sqlUtils.ts
+++ b/console/src/dataSources/services/citus/sqlUtils.ts
@ -0,0 +1,271 @@
+import { Table } from '../../types';
+
+const generateWhereClause = (
+  options: { schemas: string[]; tables: Table[] },
+  sqlTableName = 'ist.table_name',
+  sqlSchemaName = 'ist.table_schema',
+  clausePrefix = 'where'
+) => {
+  let whereClause = '';
+
+  const whereCondtions: string[] = [];
+  if (options.schemas) {
+    options.schemas.forEach(schemaName => {
+      whereCondtions.push(`(${sqlSchemaName}='${schemaName}')`);
+    });
+  }
+  if (options.tables) {
+    options.tables.forEach(tableInfo => {
+      whereCondtions.push(
+        `(${sqlSchemaName}='${tableInfo.table_schema}' and ${sqlTableName}='${tableInfo.table_name}')`
+      );
+    });
+  }
+
+  if (whereCondtions.length > 0) {
+    whereClause = clausePrefix;
+  }
+
+  whereCondtions.forEach((whereInfo, index) => {
+    whereClause += ` ${whereInfo}`;
+    if (index + 1 !== whereCondtions.length) {
+      whereClause += ' or';
+    }
+  });
+
+  return whereClause;
+};
+
+export const getFetchTablesListQuery = (options: {
+  schemas: string[];
+  tables: Table[];
+}) => {
+  const whereQuery = generateWhereClause(
+    options,
+    'pgc.relname',
+    'pgn.nspname',
+    'and'
+  );
+  return `
+  SELECT
+    COALESCE(Json_agg(Row_to_json(info)), '[]' :: json) AS tables
+  FROM (
+    with shards as (select array(select pgc.relname || '_' || pgs.shardid from pg_dist_shard as pgs inner join pg_class as pgc on pgc.oid = pgs.logicalrelid) as names),
+    partitions as (
+      select array(
+        WITH partitioned_tables AS (SELECT array(SELECT oid FROM pg_class WHERE relkind = 'p') AS parent_tables)
+        SELECT
+        child.relname       AS partition
+    FROM partitioned_tables, pg_inherits
+        JOIN pg_class child             ON pg_inherits.inhrelid   = child.oid
+        JOIN pg_namespace nmsp_child    ON nmsp_child.oid   = child.relnamespace
+    ${generateWhereClause(
+      options,
+      'child.relname',
+      'nmsp_child.nspname',
+      'where'
+    )}
+    AND pg_inherits.inhparent = ANY (partitioned_tables.parent_tables)
+      ) as names
+      )
+    SELECT
+      pgn.nspname as table_schema,
+      pgc.relname as table_name,
+      case
+        when pgc.relkind = 'r' then 'TABLE'
+        when pgc.relkind = 'f' then 'FOREIGN TABLE'
+        when pgc.relkind = 'v' then 'VIEW'
+        when pgc.relkind = 'm' then 'MATERIALIZED VIEW'
+        when pgc.relkind = 'p' then 'PARTITIONED TABLE'
+      end as table_type,
+      obj_description(pgc.oid) AS comment,
+      COALESCE(json_agg(DISTINCT row_to_json(isc) :: jsonb || jsonb_build_object('comment', col_description(pga.attrelid, pga.attnum))) filter (WHERE isc.column_name IS NOT NULL), '[]' :: json) AS columns,
+      COALESCE(json_agg(DISTINCT row_to_json(ist) :: jsonb || jsonb_build_object('comment', obj_description(pgt.oid))) filter (WHERE ist.trigger_name IS NOT NULL), '[]' :: json) AS triggers,
+      row_to_json(isv) AS view_info,
+      case
+        when cts.citus_table_type = 'reference' then 'reference'
+        when cts.citus_table_type = 'distributed' then 'distributed' 
+        else 'local'
+      end as citus_table_type
+      FROM partitions, shards, pg_class as pgc
+      INNER JOIN pg_namespace as pgn
+        ON pgc.relnamespace = pgn.oid
+    /* columns */
+    /* This is a simplified version of how information_schema.columns was
+    ** implemented in postgres 9.5, but modified to support materialized
+    ** views.
+    */
+    LEFT JOIN citus_tables as cts
+      ON pgc.oid = cts.table_name
+    LEFT OUTER JOIN pg_attribute AS pga
+      ON pga.attrelid = pgc.oid
+    LEFT OUTER JOIN (
+      SELECT
+        nc.nspname         AS table_schema,
+        c.relname          AS table_name,
+        a.attname          AS column_name,
+        a.attnum           AS ordinal_position,
+        pg_get_expr(ad.adbin, ad.adrelid) AS column_default,
+        CASE WHEN a.attnotnull OR (t.typtype = 'd' AND t.typnotnull) THEN 'NO' ELSE 'YES' END AS is_nullable,
+        CASE WHEN t.typtype = 'd' THEN
+          CASE WHEN bt.typelem <> 0 AND bt.typlen = -1 THEN 'ARRAY'
+               WHEN nbt.nspname = 'pg_catalog' THEN format_type(t.typbasetype, null)
+               ELSE 'USER-DEFINED' END
+        ELSE
+          CASE WHEN t.typelem <> 0 AND t.typlen = -1 THEN 'ARRAY'
+               WHEN nt.nspname = 'pg_catalog' THEN format_type(a.atttypid, null)
+               ELSE 'USER-DEFINED' END
+        END AS data_type,
+        coalesce(bt.typname, t.typname) AS data_type_name
+      FROM (pg_attribute a LEFT JOIN pg_attrdef ad ON attrelid = adrelid AND attnum = adnum)
+        JOIN (pg_class c JOIN pg_namespace nc ON (c.relnamespace = nc.oid)) ON a.attrelid = c.oid
+        JOIN (pg_type t JOIN pg_namespace nt ON (t.typnamespace = nt.oid)) ON a.atttypid = t.oid
+        LEFT JOIN (pg_type bt JOIN pg_namespace nbt ON (bt.typnamespace = nbt.oid))
+          ON (t.typtype = 'd' AND t.typbasetype = bt.oid)
+        LEFT JOIN (pg_collation co JOIN pg_namespace nco ON (co.collnamespace = nco.oid))
+          ON a.attcollation = co.oid AND (nco.nspname, co.collname) <> ('pg_catalog', 'default')
+      WHERE (NOT pg_is_other_temp_schema(nc.oid))
+        AND a.attnum > 0 AND NOT a.attisdropped AND c.relkind in ('r', 'v', 'm', 'f', 'p')
+        AND (pg_has_role(c.relowner, 'USAGE')
+             OR has_column_privilege(c.oid, a.attnum,
+                                     'SELECT, INSERT, UPDATE, REFERENCES'))
+    ) AS isc
+      ON  isc.table_schema = pgn.nspname
+      AND isc.table_name   = pgc.relname
+      AND isc.column_name  = pga.attname
+  
+    /* triggers */
+    LEFT OUTER JOIN pg_trigger AS pgt
+      ON pgt.tgrelid = pgc.oid
+    LEFT OUTER JOIN information_schema.triggers AS ist
+      ON  ist.event_object_schema = pgn.nspname
+      AND ist.event_object_table  = pgc.relname
+      AND ist.trigger_name        = pgt.tgname
+  
+    /* This is a simplified version of how information_schema.views was
+    ** implemented in postgres 9.5, but modified to support materialized
+    ** views.
+    */
+    LEFT OUTER JOIN (
+      SELECT
+        nc.nspname         AS table_schema,
+        c.relname          AS table_name,
+        CASE WHEN pg_has_role(c.relowner, 'USAGE') THEN pg_get_viewdef(c.oid) ELSE null END AS view_definition,
+        CASE WHEN pg_relation_is_updatable(c.oid, false) & 20 = 20 THEN 'YES' ELSE 'NO' END AS is_updatable,
+        CASE WHEN pg_relation_is_updatable(c.oid, false) &  8 =  8 THEN 'YES' ELSE 'NO' END AS is_insertable_into,
+        CASE WHEN EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = c.oid AND tgtype & 81 = 81) THEN 'YES' ELSE 'NO' END AS is_trigger_updatable,
+        CASE WHEN EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = c.oid AND tgtype & 73 = 73) THEN 'YES' ELSE 'NO' END AS is_trigger_deletable,
+        CASE WHEN EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = c.oid AND tgtype & 69 = 69) THEN 'YES' ELSE 'NO' END AS is_trigger_insertable_into
+      FROM pg_namespace nc, pg_class c
+      WHERE c.relnamespace = nc.oid
+        AND c.relkind in ('v', 'm')
+        AND (NOT pg_is_other_temp_schema(nc.oid))
+        AND (pg_has_role(c.relowner, 'USAGE')
+             OR has_table_privilege(c.oid, 'SELECT, INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER')
+             OR has_any_column_privilege(c.oid, 'SELECT, INSERT, UPDATE, REFERENCES'))
+    ) AS isv
+      ON  isv.table_schema = pgn.nspname
+      AND isv.table_name   = pgc.relname
+  
+    WHERE
+      pgc.relkind IN ('r', 'v', 'f', 'm', 'p')
+      and NOT (pgc.relname = ANY (partitions.names)) 
+      and NOT (pgc.relname = ANY (shards.names))
+      and pgc.relname != 'citus_tables'
+      ${whereQuery}      
+    GROUP BY pgc.oid, pgn.nspname, pgc.relname, table_type, isv.*, cts.citus_table_type
+  ) AS info;
+  `;
+};
+
+export const schemaListSql = (
+  schemas?: string[]
+) => `SELECT schema_name FROM information_schema.schemata WHERE
+schema_name NOT IN ('information_schema', 'pg_catalog', 'hdb_catalog', 'hdb_views', 'pg_temp_1', 'pg_toast_temp_1', 'pg_toast', 'columnar', 'citus', 'citus_internal')
+${schemas?.length ? ` AND schema_name IN (${schemas.join(',')})` : ''}
+ORDER BY schema_name ASC;`;
+
+const trackableFunctionsWhere = `
+AND has_variadic = FALSE
+AND returns_set = TRUE
+AND return_type_type = 'c'
+`;
+
+const nonTrackableFunctionsWhere = `
+AND NOT (
+  has_variadic = false
+  AND returns_set = TRUE
+  AND return_type_type = 'c'
+)
+`;
+
+const functionWhereStatement = {
+  trackable: trackableFunctionsWhere,
+  'non-trackable': nonTrackableFunctionsWhere,
+};
+
+export const getFunctionDefinitionSql = (
+  schemaName: string,
+  functionName?: string | null,
+  type?: keyof typeof functionWhereStatement
+) => `
+SELECT
+COALESCE(
+  json_agg(
+    Row_to_json(functions)
+  ),
+  '[]' :: JSON
+) from (
+SELECT * FROM (
+SELECT p.proname::text AS function_name,
+pn.nspname::text AS function_schema,
+pd.description,
+    CASE
+        WHEN p.provariadic = 0::oid THEN false
+        ELSE true
+    END AS has_variadic,
+    CASE
+        WHEN p.provolatile::text = 'i'::character(1)::text THEN 'IMMUTABLE'::text
+        WHEN p.provolatile::text = 's'::character(1)::text THEN 'STABLE'::text
+        WHEN p.provolatile::text = 'v'::character(1)::text THEN 'VOLATILE'::text
+        ELSE NULL::text
+    END AS function_type,
+pg_get_functiondef(p.oid) AS function_definition,
+rtn.nspname::text AS return_type_schema,
+rt.typname::text AS return_type_name,
+rt.typtype::text AS return_type_type,
+p.proretset AS returns_set,
+( SELECT COALESCE(json_agg(json_build_object('schema', q.schema, 'name', q.name, 'type', q.type)), '[]'::json) AS "coalesce"
+       FROM ( SELECT pt.typname AS name,
+                pns.nspname AS schema,
+                pt.typtype AS type,
+                pat.ordinality
+               FROM unnest(COALESCE(p.proallargtypes, p.proargtypes::oid[])) WITH ORDINALITY pat(oid, ordinality)
+                 LEFT JOIN pg_type pt ON pt.oid = pat.oid
+                 LEFT JOIN pg_namespace pns ON pt.typnamespace = pns.oid
+              ORDER BY pat.ordinality) q) AS input_arg_types,
+to_json(COALESCE(p.proargnames, ARRAY[]::text[])) AS input_arg_names,
+p.pronargdefaults AS default_args,
+p.oid::integer AS function_oid
+FROM pg_proc p
+ JOIN pg_namespace pn ON pn.oid = p.pronamespace
+JOIN pg_type rt ON rt.oid = p.prorettype
+JOIN pg_namespace rtn ON rtn.oid = rt.typnamespace
+JOIN pg_language plang ON p.prolang = plang.oid
+LEFT JOIN pg_description pd ON p.oid = pd.objoid
+WHERE
+plang.lanname = 'plpgsql' AND 
+pn.nspname::text !~~ 'pg_%'::text
+AND(pn.nspname::text <> ALL (ARRAY ['information_schema'::text]))
+AND NOT(EXISTS (
+  SELECT
+    1 FROM pg_aggregate
+  WHERE
+    pg_aggregate.aggfnoid::oid = p.oid))) as info
+WHERE function_schema='${schemaName}'
+${functionName ? `AND function_name='${functionName}'` : ''}
+${type ? functionWhereStatement[type] : ''}
+ORDER BY function_name ASC
+${functionName ? 'LIMIT 1' : ''}
+) as functions;
+`;
--- a/console/src/dataSources/services/citus/types.ts
+++ b/console/src/dataSources/services/citus/types.ts
@ -0,0 +1,5 @@
+import { PostgresTable } from '../postgresql/types';
+
+export interface CitusTable extends PostgresTable {
+  citus_table_type: string;
+}
--- a/console/src/dataSources/services/citus/utils.ts
+++ b/console/src/dataSources/services/citus/utils.ts
@ -0,0 +1,578 @@
+import { QualifiedTable, TableConfig } from './../../../metadata/types';
+import Endpoints from '../../../Endpoints';
+import { ReduxState } from '../../../types';
+import {
+  Relationship,
+  generateInsertRequestType,
+  RelType,
+  GenerateDeleteRowRequest,
+  GenerateBulkDeleteRowRequest,
+  BaseTableColumn,
+} from '../../types';
+import { isEmpty } from '../../../components/Common/utils/jsUtils';
+import { CitusTable } from './types';
+import {
+  getColQuery,
+  getFullQueryNameBase,
+  getGraphQLQueryBase,
+} from '../../common';
+import { WhereClause } from '../../../components/Common/utils/v1QueryUtils';
+
+type Tables = ReduxState['tables'];
+
+interface QueryBody {
+  clauses: string;
+  relationshipInfo: Relationship[];
+}
+
+export const CitusDataTypes = {
+  character: [
+    'char',
+    'varchar',
+    'text',
+    'nchar',
+    'nvarchar',
+    'binary',
+    'vbinary',
+    'image',
+    'character',
+    'citext',
+  ],
+  numeric: [
+    'smallint',
+    'integer',
+    'double precision',
+    'int',
+    'int2',
+    'int4',
+    'int8',
+    'bigint',
+    'decimal',
+    'numeric',
+    'smallmoney',
+    'money',
+    'float',
+    'real',
+  ],
+  dateTime: [
+    'datetime',
+    'smalldatetime',
+    'date',
+    'time',
+    'datetimeoffset',
+    'timestamp',
+  ],
+  user_defined: [],
+};
+
+export const operators = [
+  { name: 'equals', value: '$eq', graphqlOp: '_eq' },
+  { name: 'not equals', value: '$ne', graphqlOp: '_neq' },
+  { name: '>', value: '$gt', graphqlOp: '_gt' },
+  { name: '<', value: '$lt', graphqlOp: '_lt' },
+  { name: '>=', value: '$gte', graphqlOp: '_gte' },
+  { name: '<=', value: '$lte', graphqlOp: '_lte' },
+];
+
+const getFormattedValue = (
+  type: string,
+  value: any
+): string | number | undefined => {
+  if (
+    CitusDataTypes.character.includes(type) ||
+    CitusDataTypes.dateTime.includes(type)
+  )
+    return `"${value}"`;
+
+  if (CitusDataTypes.numeric.includes(type)) return value;
+};
+
+const getFullQueryName = getFullQueryNameBase('public');
+
+export const getRowsCountRequestBody = ({
+  tables,
+  tableConfiguration,
+}: {
+  tables: Tables;
+  tableConfiguration: TableConfig;
+}) => {
+  const {
+    currentTable: originalTable,
+    currentSchema,
+    allSchemas,
+    view,
+  } = tables;
+  const queryBody = ({ clauses }: QueryBody) => {
+    const queryName = getFullQueryName({
+      tableName: originalTable,
+      schema: currentSchema,
+      tableConfiguration,
+      operation: 'select_aggregate',
+    });
+    return `query TableCount {
+      ${queryName} ${clauses && `(${clauses})`} {
+        aggregate {
+          count
+        }
+      }
+    }`;
+  };
+
+  return {
+    query: getGraphQLQueryBase({
+      allSchemas,
+      view,
+      originalTable,
+      currentSchema,
+      isExport: true,
+      tableConfiguration,
+      queryBody,
+      getFormattedValue,
+    }),
+    variables: null,
+    operationName: 'TableCount',
+  };
+};
+
+const processCount = (c: {
+  data: any;
+  currentSchema: string;
+  originalTable: string;
+  tableConfiguration: TableConfig;
+}): number => {
+  const key = getFullQueryName({
+    tableName: c.originalTable,
+    schema: c.currentSchema,
+    tableConfiguration: c.tableConfiguration,
+    operation: 'select_aggregate',
+  });
+  return c.data?.data[key]?.aggregate?.count;
+};
+
+export const generateRowsCountRequest = () => ({
+  getRowsCountRequestBody,
+  endpoint: Endpoints.graphQLUrl,
+  processCount,
+});
+
+export const getTableRowRequestBody = ({
+  tables,
+  isExport,
+  tableConfiguration,
+}: {
+  tables: Tables;
+  isExport?: boolean;
+  tableConfiguration: TableConfig;
+}) => {
+  const {
+    currentTable: originalTable,
+    view,
+    allSchemas,
+    currentSchema,
+  } = tables;
+  const queryName = getFullQueryName({
+    tableName: originalTable,
+    schema: currentSchema,
+    tableConfiguration,
+    operation: 'select',
+  });
+  const aggregateName = getFullQueryName({
+    tableName: originalTable,
+    schema: currentSchema,
+    tableConfiguration,
+    operation: 'select_aggregate',
+  });
+  const queryBody = ({ clauses, relationshipInfo }: QueryBody) => {
+    return `query TableRows {
+      ${queryName} ${clauses && `(${clauses})`} {
+          ${getColQuery(
+            view.query.columns,
+            view.curFilter.limit,
+            relationshipInfo,
+            tableConfiguration
+          ).join('\n')} 
+    }
+    ${aggregateName} {
+      aggregate {
+        count
+      }
+    }
+  }`;
+  };
+
+  return {
+    query: getGraphQLQueryBase({
+      allSchemas,
+      view,
+      originalTable,
+      currentSchema,
+      isExport,
+      tableConfiguration,
+      queryBody,
+      getFormattedValue,
+    }),
+    variables: null,
+    operationName: 'TableRows',
+  };
+};
+
+export const processTableRowData = (
+  data: any,
+  config: {
+    originalTable: string;
+    currentSchema: string;
+    tableConfiguration: TableConfig;
+  }
+) => {
+  try {
+    const { originalTable, currentSchema, tableConfiguration } = config!;
+
+    const reversedCustomColumns = Object.entries(
+      tableConfiguration?.custom_column_names ?? {}
+    ).reduce((acc: Record<string, string>, [col, customCol]) => {
+      acc[customCol] = col;
+      return acc;
+    }, {});
+
+    const queryName = getFullQueryName({
+      tableName: originalTable,
+      schema: currentSchema,
+      tableConfiguration,
+      operation: 'select',
+    });
+    const results = data?.data[queryName];
+
+    const rows = isEmpty(reversedCustomColumns)
+      ? results
+      : results?.map((row: Record<string, any>) =>
+          Object.entries(row).reduce(
+            (acc: Record<string, any>, [maybeCustomCol, col]) => {
+              acc[
+                reversedCustomColumns[maybeCustomCol] ?? maybeCustomCol
+              ] = col;
+              return acc;
+            },
+            {}
+          )
+        );
+    const aggregateName = getFullQueryName({
+      tableName: originalTable,
+      schema: currentSchema,
+      tableConfiguration,
+      operation: 'select_aggregate',
+    });
+    const estimatedCount =
+      data?.data[aggregateName]?.aggregate?.count ?? rows.length;
+    return { estimatedCount, rows };
+  } catch (err) {
+    throw new Error('data source is inconsistent');
+  }
+};
+
+export const generateTableRowRequest = () => ({
+  endpoint: Endpoints.graphQLUrl,
+  getTableRowRequestBody,
+  processTableRowData,
+});
+
+const getInsertRequestBody = (
+  data: Parameters<generateInsertRequestType['getInsertRequestBody']>[0]
+): ReturnType<generateInsertRequestType['getInsertRequestBody']> => {
+  const { name: tableName, schema } = data.tableDef;
+  const { tableConfiguration } = data;
+  const customColumnNames = tableConfiguration?.custom_column_names ?? {};
+
+  const processedData: Record<string, any> = {};
+  Object.entries(data.insertObject).forEach(([key, value]) => {
+    processedData[customColumnNames[key] || key] = value;
+  });
+  const values = Object.entries(processedData).map(([key, value]) => {
+    return `${key}: ${typeof value === 'string' ? `"${value}"` : value}`;
+  });
+  const returning = Object.keys(processedData).join('\n');
+
+  const queryName = getFullQueryName({
+    tableName,
+    schema,
+    tableConfiguration,
+    operation: 'insert',
+  });
+
+  const query = `
+  mutation InsertRow {
+     ${queryName}(objects: { ${values} }){
+         returning { 
+           ${returning}
+         }
+     }
+  }
+  `;
+
+  return {
+    query,
+    variables: null,
+  };
+};
+
+type processInsertDataParameter = Parameters<
+  generateInsertRequestType['processInsertData']
+>;
+
+const processInsertData = (
+  result: processInsertDataParameter[0],
+  tableConfiguration: TableConfig,
+  config: processInsertDataParameter[2]
+) => {
+  const { currentTable, currentSchema } = config!;
+  const index = getFullQueryName({
+    tableName: currentTable,
+    schema: currentSchema,
+    tableConfiguration,
+    operation: 'insert',
+  });
+  const returnedFields = (result as {
+    data: Record<string, Record<string, any>>;
+  })?.data?.[index]?.returning;
+  return {
+    affectedRows: 1,
+    returnedFields:
+      returnedFields?.length === 1 ? returnedFields[0] : returnedFields,
+  };
+};
+
+export const generateInsertRequest = () => ({
+  endpoint: Endpoints.graphQLUrl,
+  processInsertData,
+  getInsertRequestBody,
+});
+
+export const isRelationshipValid = (
+  rel: RelType,
+  lTable: CitusTable,
+  rTable: CitusTable
+) => {
+  if (rel.isObjRel) {
+    if (
+      ['reference', 'local'].includes(lTable.citus_table_type) &&
+      rTable.citus_table_type === 'distributed'
+    )
+      return false;
+
+    if (
+      lTable.citus_table_type === 'distributed' &&
+      rTable.citus_table_type === 'local'
+    )
+      return false;
+
+    return true;
+  }
+
+  /* Array relationship rules */
+  if (
+    ['reference', 'local'].includes(lTable.citus_table_type) &&
+    rTable.citus_table_type === 'distributed'
+  )
+    return false;
+
+  if (
+    lTable.citus_table_type === 'distributed' &&
+    ['reference', 'local'].includes(rTable.citus_table_type)
+  )
+    return false;
+
+  return true;
+};
+
+const getEditRowRequestBody = (data: {
+  source: string;
+  tableDef: QualifiedTable;
+  tableConfiguration: TableConfig;
+  set: Record<string, any>;
+  where: Record<string, any>;
+}) => {
+  const { tableConfiguration } = data;
+  const customColumnNames = tableConfiguration?.custom_column_names || {};
+  const whereClause = Object.entries(data.where)
+    .map(
+      ([key, value]) =>
+        `${customColumnNames[key] || key}: {_eq: ${
+          typeof value === 'string' ? `"${value}"` : value
+        }}`
+    )
+    .join(', ');
+
+  const setClause = Object.entries(data.set)
+    .map(
+      ([key, value]) =>
+        `${customColumnNames[key] || key}: ${
+          typeof value === 'string' ? `"${value}"` : value
+        }`
+    )
+    .join(', ');
+  const { name: tableName, schema } = data.tableDef;
+  const operationName = getFullQueryName({
+    tableName,
+    schema,
+    tableConfiguration,
+    operation: 'update',
+  });
+
+  const query = `mutation {
+    ${operationName}(where: {${whereClause}}, _set: {${setClause}}) {
+      affected_rows
+    }
+  }`;
+  return {
+    query,
+    variables: null,
+  };
+};
+
+const processEditData = ({
+  data,
+  tableDef,
+  tableConfiguration,
+}: {
+  tableDef: QualifiedTable;
+  data: any;
+  tableConfiguration: TableConfig;
+}): number => {
+  const { name: tableName, schema } = tableDef;
+  const operationName = getFullQueryName({
+    tableName,
+    schema,
+    tableConfiguration,
+    operation: 'update',
+  });
+  return data?.data[operationName].affected_rows;
+};
+
+export const generateEditRowRequest = () => ({
+  endpoint: Endpoints.graphQLUrl,
+  processEditData,
+  getEditRowRequestBody,
+});
+
+const getDeleteRowRequestBody = ({
+  pkClause,
+  tableName,
+  schemaName,
+  columnInfo,
+  tableConfiguration,
+}: {
+  pkClause: WhereClause;
+  tableName: string;
+  schemaName: string;
+  columnInfo: BaseTableColumn[];
+  tableConfiguration: TableConfig;
+}) => {
+  const customColumns = tableConfiguration?.custom_column_names;
+
+  const args = Object.keys(pkClause)
+    .map(key => {
+      let value = (pkClause as Record<string, any>)[key];
+      const column = columnInfo.find(c => c.column_name === key);
+      const columnName =
+        customColumns && customColumns[key] ? customColumns[key] : key;
+      value = getFormattedValue(column?.data_type_name || 'varchar', value);
+      return `${columnName}: {_eq: ${value}}`;
+    })
+    .join(',');
+  const identifier = getFullQueryName({
+    tableName,
+    schema: schemaName,
+    tableConfiguration,
+    operation: 'delete',
+  });
+  const query = `mutation DeleteRows {
+    delete_row: ${identifier}(where: {${args}}) {
+      affected_rows
+    }
+  }`;
+  return {
+    query,
+    variables: null,
+  };
+};
+
+const processDeleteRowData = (data: Record<string, any>) => {
+  try {
+    if (data.errors) throw new Error(data.errors[0].message);
+    if (data?.data?.delete_row?.affected_rows)
+      return data?.data?.delete_row?.affected_rows;
+    throw new Error('Invalid response');
+  } catch (err) {
+    throw new Error(err.message);
+  }
+};
+
+export const generateDeleteRowRequest = (): GenerateDeleteRowRequest => ({
+  endpoint: Endpoints.graphQLUrl,
+  getDeleteRowRequestBody,
+  processDeleteRowData,
+});
+
+const getBulkDeleteRowRequestBody = ({
+  pkClauses,
+  tableName,
+  schemaName,
+  columnInfo,
+  tableConfiguration,
+}: {
+  pkClauses: WhereClause[];
+  tableName: string;
+  schemaName: string;
+  columnInfo: BaseTableColumn[];
+  tableConfiguration: TableConfig;
+}) => {
+  const customColumns = tableConfiguration?.custom_column_names;
+  const identifier = getFullQueryName({
+    tableName,
+    schema: schemaName,
+    tableConfiguration,
+    operation: 'delete',
+  });
+  const topLevelFields = pkClauses.map((pkClause, i) => {
+    const args = Object.keys(pkClause)
+      .map(key => {
+        let value = (pkClause as Record<string, any>)[key];
+        const column = columnInfo.find(c => c.column_name === key);
+        const columnName =
+          customColumns && customColumns[key] ? customColumns[key] : key;
+        value = getFormattedValue(column?.data_type_name || 'varchar', value);
+        return `${columnName}: {_eq: ${value}}`;
+      })
+      .join(',');
+
+    return `delete_row_${i}: ${identifier}(where: {${args}}) { affected_rows }`;
+  });
+  const query = `mutation MyMutation {
+    ${topLevelFields.join('\n')}
+  }`;
+  return {
+    query,
+    variables: null,
+  };
+};
+
+const processBulkDeleteRowData = (data: Record<string, any>) => {
+  try {
+    if (data.errors) throw new Error(data.errors[0].message);
+
+    if (data.data) {
+      const res = Object.keys(data.data)
+        .filter(key => data.data[key])
+        .map(key => data.data[key].affected_rows)
+        .reduce((a, b) => a + b, 0);
+      return res;
+    }
+  } catch (err) {
+    throw new Error(err.message);
+  }
+};
+
+export const generateBulkDeleteRowRequest = (): GenerateBulkDeleteRowRequest => ({
+  endpoint: Endpoints.graphQLUrl,
+  getBulkDeleteRowRequestBody,
+  processBulkDeleteRowData,
+});
--- a/console/src/dataSources/services/index.ts
+++ b/console/src/dataSources/services/index.ts
@ -2,5 +2,6 @@ import { postgres } from './postgresql';
 import { mysql } from './mysql';
 import { mssql } from './mssql';
 import { bigquery } from './bigquery';
+import { citus } from './citus';

-export const services = { postgres, mysql, mssql, bigquery };
+export const services = { postgres, mysql, mssql, bigquery, citus };
--- a/console/src/dataSources/services/mssql/index.tsx
+++ b/console/src/dataSources/services/mssql/index.tsx
@ -9,7 +9,11 @@ import {
  FrequentlyUsedColumn,
  ViolationActions,
 } from '../../types';
-import { generateTableRowRequest, operators } from './utils';
+import {
+  generateTableRowRequest,
+  operators,
+  generateRowsCountRequest,
+} from './utils';

 const permissionColumnDataTypes = {
  character: [
@ -77,7 +81,7 @@ const columnDataTypes = {
 };

 // eslint-disable-next-line no-useless-escape
-const createSQLRegex = /create\s*(?:|or\s*replace)\s*(?<type>view|table|function)\s*(?:\s*if*\s*not\s*exists\s*)?((?<schema>\"?\w+\"?)\.(?<tableWithSchema>\"?\w+\"?)|(?<table>\"?\w+\"?))\s*(?<partition>partition\s*of)?/gim;
+const createSQLRegex = /create\s*(?:|or\s*replace)\s*(?<type>view|table|function)\s*(?:\s*if*\s*not\s*exists\s*)?((?<schema>\"?\w+\"?)\.(?<nameWithSchema>\"?\w+\"?)|(?<name>\"?\w+\"?))\s*(?<partition>partition\s*of)?/gim;

 export const displayTableName = (table: Table) => {
  const tableName = table.table_name;
@ -112,7 +116,7 @@ export const supportedFeatures: SupportedFeaturesType = {
    browse: {
      enabled: true,
      customPagination: true,
-      aggregation: false,
+      aggregation: true,
    },
    insert: {
      enabled: false,
@ -899,4 +903,5 @@ WHERE
  supportedFeatures,
  violationActions,
  defaultRedirectSchema,
+  generateRowsCountRequest,
 };
--- a/console/src/dataSources/services/mssql/utils.ts
+++ b/console/src/dataSources/services/mssql/utils.ts
@ -1,4 +1,9 @@
-import { TableEntry } from './../../../metadata/types';
+import {
+  getFullQueryNameBase,
+  getGraphQLQueryBase,
+  QueryBody,
+} from './../../common';
+import { TableConfig } from './../../../metadata/types';
 import {
  OrderBy,
  WhereClause,
@ -15,7 +20,8 @@ interface GetGraphQLQuery {
  originalTable: string;
  currentSchema: string;
  isExport?: boolean;
-  tableConfiguration: TableEntry['configuration'];
+  tableConfiguration: TableConfig;
+  defaultSchema: string;
 }

 export const SQLServerTypes = {
@ -62,6 +68,8 @@ export const operators = [
  { name: '<=', value: '$lte', graphqlOp: '_lte' },
 ];

+const getFullQueryName = getFullQueryNameBase('dbo');
+
 const getFormattedValue = (
  type: string,
  value: any
@ -85,7 +93,7 @@ const RqlToGraphQlOp = (op: string) => {
 const generateWhereClauseQueryString = (
  wheres: WhereClause[],
  columnTypeInfo: BaseTableColumn[],
-  tableConfiguration: TableEntry['configuration']
+  tableConfiguration: TableConfig
 ): string | null => {
  const customColumns = tableConfiguration?.custom_column_names ?? {};
  const whereClausesArr = wheres.map((i: Record<string, any>) => {
@ -105,7 +113,7 @@ const generateWhereClauseQueryString = (

 const generateSortClauseQueryString = (
  sorts: OrderBy[],
-  tableConfiguration: TableEntry['configuration']
+  tableConfiguration: TableConfig
 ): string | null => {
  const customColumns = tableConfiguration?.custom_column_names ?? {};
  const sortClausesArr = sorts.map((i: OrderBy) => {
@ -120,7 +128,7 @@ const getColQuery = (
  cols: (string | { name: string; columns: string[] })[],
  limit: number,
  relationships: Relationship[],
-  tableConfiguration: TableEntry['configuration']
+  tableConfiguration: TableConfig
 ): string[] => {
  return cols.map(c => {
    const customColumns = tableConfiguration?.custom_column_names ?? {};
@ -142,6 +150,7 @@ export const getGraphQLQueryForBrowseRows = ({
  currentSchema,
  isExport = false,
  tableConfiguration,
+  defaultSchema,
 }: GetGraphQLQuery) => {
  const currentTable: Table | undefined = allSchemas?.find(
    (t: Table) =>
@ -190,10 +199,12 @@ export const getGraphQLQueryForBrowseRows = ({
  ]
    .filter(Boolean)
    .join(',')}`;
-  const tableName = tableConfiguration?.custom_name || originalTable;
+
  return `query TableRows {
      ${
-        currentSchema === 'dbo' ? tableName : `${currentSchema}_${tableName}`
+        currentSchema === defaultSchema
+          ? originalTable
+          : `${currentSchema}_${originalTable}`
      } ${clauses && `(${clauses})`} {
          ${getColQuery(
            view.query.columns,
@ -212,24 +223,55 @@ export const getTableRowRequestBody = ({
 }: {
  tables: Tables;
  isExport?: boolean;
-  tableConfiguration?: TableEntry['configuration'];
+  tableConfiguration: TableConfig;
 }) => {
-  // TODO: fetch count when agregation for mssql is added
  const {
    currentTable: originalTable,
    view,
    allSchemas,
    currentSchema,
  } = tables;
+  const tableName = tableConfiguration?.custom_name ?? originalTable;
+  const queryName = getFullQueryName({
+    tableName,
+    schema: currentSchema,
+    tableConfiguration,
+    operation: 'select',
+  });
+  const aggregateName = getFullQueryName({
+    tableName,
+    schema: currentSchema,
+    tableConfiguration,
+    operation: 'select_aggregate',
+  });
+  const queryBody = ({ clauses, relationshipInfo }: QueryBody) => {
+    return `query TableRows {
+      ${queryName} ${clauses && `(${clauses})`} {
+          ${getColQuery(
+            view.query.columns,
+            view.curFilter.limit,
+            relationshipInfo,
+            tableConfiguration
+          ).join('\n')} 
+    }
+    ${aggregateName} {
+      aggregate {
+        count
+      }
+    }
+  }`;
+  };

  return {
-    query: getGraphQLQueryForBrowseRows({
+    query: getGraphQLQueryBase({
      allSchemas,
      view,
      originalTable,
      currentSchema,
      isExport,
      tableConfiguration,
+      queryBody,
+      getFormattedValue,
    }),
    variables: null,
    operationName: 'TableRows',
@ -241,24 +283,26 @@ const processTableRowData = (
  config?: {
    originalTable: string;
    currentSchema: string;
-    tableConfiguration?: TableEntry['configuration'];
+    tableConfiguration: TableConfig;
  }
 ) => {
-  const { originalTable, currentSchema } = config!;
+  const { originalTable, currentSchema, tableConfiguration } = config!;

  const reversedCustomColumns = Object.entries(
-    config?.tableConfiguration?.custom_column_names ?? {}
+    tableConfiguration?.custom_column_names ?? {}
  ).reduce((acc: Record<string, string>, [col, customCol]) => {
    acc[customCol] = col;
    return acc;
  }, {});

-  const tableConfiguration = config?.tableConfiguration;
  const tableName = tableConfiguration?.custom_name || originalTable;
-  const results =
-    data.data[
-      currentSchema === 'dbo' ? tableName : `${currentSchema}_${tableName}`
-    ];
+  const queryName = getFullQueryName({
+    tableName,
+    schema: currentSchema,
+    tableConfiguration,
+    operation: 'select',
+  });
+  const results = data?.data[queryName];

  const rows = isEmpty(reversedCustomColumns)
    ? results
@ -271,13 +315,79 @@ const processTableRowData = (
          {}
        )
      );
-  return { estimatedCount: rows.length, rows };
+  const estimatedCount =
+    data?.data[`${queryName}_aggregate`]?.aggregate?.count ?? rows.length;
+  return { estimatedCount, rows };
 };

-export const generateTableRowRequest = () => {
+export const generateTableRowRequest = () => ({
+  endpoint: Endpoints.graphQLUrl,
+  getTableRowRequestBody,
+  processTableRowData,
+});
+
+export const getRowsCountRequestBody = ({
+  tables,
+  tableConfiguration,
+}: {
+  tables: Tables;
+  tableConfiguration: TableConfig;
+}) => {
+  const {
+    currentTable: originalTable,
+    currentSchema,
+    allSchemas,
+    view,
+  } = tables;
+  const queryBody = ({ clauses }: QueryBody) => {
+    const queryName = getFullQueryName({
+      tableName: originalTable,
+      schema: currentSchema,
+      tableConfiguration,
+      operation: 'select_aggregate',
+    });
+    return `query TableCount {
+      ${queryName} ${clauses && `(${clauses})`} {
+        aggregate {
+          count
+        }
+      }
+    }`;
+  };
+
  return {
-    endpoint: Endpoints.graphQLUrl,
-    getTableRowRequestBody,
-    processTableRowData,
+    query: getGraphQLQueryBase({
+      allSchemas,
+      view,
+      originalTable,
+      currentSchema,
+      isExport: true,
+      tableConfiguration,
+      queryBody,
+      getFormattedValue,
+    }),
+    variables: null,
+    operationName: 'TableCount',
  };
 };
+
+const processCount = (c: {
+  data: any;
+  currentSchema: string;
+  originalTable: string;
+  tableConfiguration: TableConfig;
+}): number => {
+  const key = getFullQueryName({
+    tableName: c.originalTable,
+    schema: c.currentSchema,
+    tableConfiguration: c.tableConfiguration,
+    operation: 'select_aggregate',
+  });
+  return c.data?.data?.[key]?.aggregate?.count;
+};
+
+export const generateRowsCountRequest = () => ({
+  getRowsCountRequestBody,
+  endpoint: Endpoints.graphQLUrl,
+  processCount,
+});
--- a/console/src/dataSources/services/postgresql/index.tsx
+++ b/console/src/dataSources/services/postgresql/index.tsx
@ -10,7 +10,14 @@ import {
 import { QUERY_TYPES, Operations } from '../../common';
 import { PGFunction } from './types';
 import { DataSourcesAPI, ColumnsInfoResult } from '../..';
-import { generateTableRowRequest } from './utils';
+import {
+  generateTableRowRequest,
+  generateInsertRequest,
+  generateRowsCountRequest,
+  generateEditRowRequest,
+  generateDeleteRowRequest,
+  generateBulkDeleteRowRequest,
+} from './utils';
 import {
  getFetchTablesListQuery,
  fetchColumnTypesQuery,
@ -429,7 +436,7 @@ export const isColTypeString = (colType: string) =>

 const dependencyErrorCode = '2BP01'; // pg dependent error > https://www.postgresql.org/docs/current/errcodes-appendix.html

-const createSQLRegex = /create\s*(?:|or\s*replace)\s*(?<type>view|table|function)\s*(?:\s*if*\s*not\s*exists\s*)?((?<schema>\"?\w+\"?)\.(?<tableWithSchema>\"?\w+\"?)|(?<table>\"?\w+\"?))\s*(?<partition>partition\s*of)?/gim; // eslint-disable-line
+const createSQLRegex = /create\s*(?:|or\s*replace)\s*(?<type>view|table|function)\s*(?:\s*if*\s*not\s*exists\s*)?((?<schema>\"?\w+\"?)\.(?<nameWithSchema>\"?\w+\"?)|(?<name>\"?\w+\"?))\s*(?<partition>partition\s*of)?/gim; // eslint-disable-line

 const isTimeoutError = (error: {
  code: string;
@ -726,5 +733,10 @@ export const postgres: DataSourcesAPI = {
  supportedFeatures,
  violationActions,
  defaultRedirectSchema,
+  generateInsertRequest,
+  generateRowsCountRequest,
  getPartitionDetailsSql,
+  generateEditRowRequest,
+  generateDeleteRowRequest,
+  generateBulkDeleteRowRequest,
 };
--- a/console/src/dataSources/services/postgresql/sqlUtils.ts
+++ b/console/src/dataSources/services/postgresql/sqlUtils.ts
@ -60,7 +60,6 @@ export const getFetchTablesListQuery = (options: {
    'pgn.nspname',
    'and'
  );
-
  return `
  SELECT
    COALESCE(Json_agg(Row_to_json(info)), '[]' :: json) AS tables
@ -96,11 +95,9 @@ export const getFetchTablesListQuery = (options: {
      COALESCE(json_agg(DISTINCT row_to_json(isc) :: jsonb || jsonb_build_object('comment', col_description(pga.attrelid, pga.attnum))) filter (WHERE isc.column_name IS NOT NULL), '[]' :: json) AS columns,
      COALESCE(json_agg(DISTINCT row_to_json(ist) :: jsonb || jsonb_build_object('comment', obj_description(pgt.oid))) filter (WHERE ist.trigger_name IS NOT NULL), '[]' :: json) AS triggers,
      row_to_json(isv) AS view_info
-
-    FROM pg_class as pgc
-    INNER JOIN pg_namespace as pgn
-      ON pgc.relnamespace = pgn.oid
-
+      FROM partitions, pg_class as pgc  
+      INNER JOIN pg_namespace as pgn
+        ON pgc.relnamespace = pgn.oid
    /* columns */
    /* This is a simplified version of how information_schema.columns was
    ** implemented in postgres 9.5, but modified to support materialized
--- a/console/src/dataSources/services/postgresql/utils.ts
+++ b/console/src/dataSources/services/postgresql/utils.ts
@ -2,10 +2,18 @@ import { dataSource, generateTableDef } from '../..';
 import {
  getRunSqlQuery,
  getSelectQuery,
+  WhereClause,
 } from '../../../components/Common/utils/v1QueryUtils';
 import Endpoints, { globalCookiePolicy } from '../../../Endpoints';
 import requestAction from '../../../utils/requestAction';
+import {
+  generateInsertRequestType,
+  GenerateBulkDeleteRowRequest,
+  GenerateDeleteRowRequest,
+} from '../../types';
 import { ReduxState } from './../../../types';
+import { getStatementTimeoutSql } from './sqlUtils';
+import { QualifiedTable, TableConfig } from '../../../metadata/types';

 type Tables = ReduxState['tables'];
 type Headers = Tables['dataHeaders'];
@ -112,3 +120,189 @@ export const generateTableRowRequest = () => {
    processTableRowData,
  };
 };
+
+const getInsertRequestBody = ({
+  source,
+  tableDef,
+  insertObject,
+  returning,
+}: Parameters<
+  generateInsertRequestType['getInsertRequestBody']
+>[0]): ReturnType<generateInsertRequestType['getInsertRequestBody']> => {
+  return {
+    type: 'insert',
+    args: {
+      source,
+      table: tableDef,
+      objects: [insertObject],
+      returning,
+    },
+  };
+};
+
+type processInsertDataParameter = Parameters<
+  generateInsertRequestType['processInsertData']
+>;
+
+const processInsertData = (result: processInsertDataParameter[0]) => {
+  result = result as Record<string, Record<string, any>>;
+  return {
+    affectedRows: result?.affected_rows,
+    returnedFields: result?.returning[0],
+  };
+};
+
+export const generateInsertRequest = () => ({
+  endpoint: Endpoints.query,
+  processInsertData,
+  getInsertRequestBody,
+});
+
+export const getRowsCountRequestBody = ({
+  tables,
+}: {
+  tables: Tables;
+  tableConfiguration: TableConfig;
+}) => {
+  const {
+    currentTable: originalTable,
+    currentSchema,
+    view,
+    currentDataSource,
+  } = tables;
+  const selectQuery = getSelectQuery(
+    'count',
+    generateTableDef(originalTable, currentSchema),
+    view.query.columns,
+    view.query.where,
+    view.query.offset,
+    view.query.limit,
+    view.query.order_by,
+    currentDataSource
+  );
+
+  const queries = [
+    getRunSqlQuery(getStatementTimeoutSql(2), currentDataSource),
+    selectQuery,
+  ];
+
+  return {
+    type: 'bulk',
+    source: currentDataSource,
+    args: queries,
+  };
+};
+
+export const processCount = ({
+  data,
+}: {
+  data: any;
+  currentSchema: string;
+  originalTable: string;
+  tableConfiguration: TableConfig;
+}): number => {
+  return data[1].count;
+};
+
+export const generateRowsCountRequest = () => ({
+  getRowsCountRequestBody,
+  endpoint: Endpoints.query,
+  processCount,
+});
+
+const getEditRowRequestBody = (data: {
+  source: string;
+  tableDef: QualifiedTable;
+  set: Record<string, any>;
+  where: Record<string, any>;
+  defaultArray: any[];
+}) => {
+  return {
+    type: 'update',
+    args: {
+      source: data.source,
+      table: data.tableDef,
+      $set: data.set,
+      $default: data.defaultArray,
+      where: data.where,
+    },
+  };
+};
+
+const processEditData = ({
+  data,
+}: {
+  tableDef: QualifiedTable;
+  data: any;
+}): number => {
+  return data.affected_rows;
+};
+
+export const generateEditRowRequest = () => ({
+  endpoint: Endpoints.query,
+  processEditData,
+  getEditRowRequestBody,
+});
+
+const getDeleteRowRequestBody = ({
+  pkClause,
+  tableName,
+  schemaName,
+  source,
+}: {
+  pkClause: WhereClause;
+  tableName: string;
+  schemaName: string;
+  source: string;
+}) => {
+  return {
+    type: 'delete',
+    args: {
+      source,
+      table: {
+        name: tableName,
+        schema: schemaName,
+      },
+      where: pkClause,
+    },
+  };
+};
+
+const processDeleteRowData = (data: any) => {
+  return data.affected_rows;
+};
+
+export const generateDeleteRowRequest = (): GenerateDeleteRowRequest => ({
+  endpoint: Endpoints.query,
+  getDeleteRowRequestBody,
+  processDeleteRowData,
+});
+
+const getBulkDeleteRowRequestBody = ({
+  pkClauses,
+  tableName,
+  schemaName,
+  source,
+}: {
+  pkClauses: WhereClause[];
+  tableName: string;
+  schemaName: string;
+  source: string;
+}) => {
+  return {
+    type: 'bulk',
+    source,
+    args: pkClauses.map(pkClause =>
+      getDeleteRowRequestBody({ pkClause, tableName, schemaName, source })
+    ),
+  };
+};
+
+const processBulkDeleteRowData = (data: any) =>
+  data.reduce((acc: any, d: any) => acc + d.affected_rows, 0);
+
+export const generateBulkDeleteRowRequest = (): GenerateBulkDeleteRowRequest => ({
+  endpoint: Endpoints.query,
+  getBulkDeleteRowRequestBody,
+  processBulkDeleteRowData,
+});
--- a/console/src/dataSources/types.ts
+++ b/console/src/dataSources/types.ts
@ -2,13 +2,15 @@ import { Nullable } from '../components/Common/utils/tsUtils';
 import { Column } from '../utils/postgresColumnTypes';
 import {
  FunctionDefinition,
+  QualifiedTable,
  RemoteRelationshipDef,
-  TableEntry,
+  TableConfig,
 } from '../metadata/types';
 import { ReduxState } from '../types';
 import {
  getSelectQuery,
  getRunSqlQuery,
+  WhereClause,
 } from '../../src/components/Common/utils/v1QueryUtils';

 export interface Relationship
@ -159,6 +161,7 @@ export interface Table extends BaseTable {
    table_schema: string;
    definition: RemoteRelationshipDef;
  }[];
+  citusTableType?: string;
  unique_constraints:
    | {
        table_name: string;
@ -334,7 +337,7 @@ export type generateTableRowRequestType = {
  getTableRowRequestBody: (data: {
    tables: Tables;
    isExport?: boolean;
-    tableConfiguration?: TableEntry['configuration'];
+    tableConfiguration: TableConfig;
  }) =>
    | {
        type: string;
@ -351,14 +354,163 @@ export type generateTableRowRequestType = {
      };
  processTableRowData: <T>(
    data: T,
-    config?: {
+    config: {
      originalTable: string;
      currentSchema: string;
-      tableConfiguration?: TableEntry['configuration'];
+      tableConfiguration: TableConfig;
    }
  ) => { rows: T[]; estimatedCount: number };
 };

+export type generateInsertRequestType = {
+  endpoint: string;
+  getInsertRequestBody: (data: {
+    tableDef: QualifiedTable;
+    source: string;
+    insertObject: Record<string, any>;
+    tableConfiguration: TableConfig;
+    returning: string[];
+  }) =>
+    | {
+        type: 'insert';
+        args: {
+          source: string;
+          table: QualifiedTable;
+          objects: [Record<string, any>];
+          returning: string[];
+        };
+      }
+    | {
+        query: string;
+        variables: null;
+      };
+  processInsertData: (
+    data:
+      | { affectedRows: number; returning: Array<Record<string, any>> }
+      | Record<string, Record<string, any>>,
+    tableConfiguration: TableConfig,
+    config: {
+      currentTable: string;
+      currentSchema: string;
+    }
+  ) => {
+    affectedRows: number | Record<string, any>;
+    returnedFields: Record<string, any>;
+  };
+};
+
+export type GenerateRowsCountRequestType = {
+  endpoint: string;
+  getRowsCountRequestBody: generateTableRowRequestType['getTableRowRequestBody'];
+  processCount: (config: {
+    data: any;
+    originalTable: string;
+    currentSchema: string;
+    tableConfiguration: TableConfig;
+  }) => number;
+};
+
+export type GenerateEditRowRequest = {
+  endpoint: string;
+  processEditData: (args: {
+    tableDef: QualifiedTable;
+    tableConfiguration: TableConfig;
+    data: any;
+  }) => number;
+  getEditRowRequestBody: (data: {
+    source: string;
+    tableDef: QualifiedTable;
+    tableConfiguration: TableConfig;
+    set: Record<string, any>;
+    where: Record<string, any>;
+    defaultArray: any[];
+  }) =>
+    | {
+        type: string;
+        args: {
+          source: string;
+          table: QualifiedTable;
+          $set: Record<string, any>;
+          $default: any[];
+          where: Record<string, any>;
+        };
+      }
+    | {
+        query: string;
+        variables: null;
+      };
+};
+
+export type RelType = {
+  relName: string;
+  lTable: string;
+  lSchema: string;
+  isObjRel: boolean;
+  lcol: string[] | null;
+  rcol: string[] | null;
+  rTable: string | null;
+  rSchema: string | null;
+};
+
+export type GenerateDeleteRowRequest = {
+  endpoint: string;
+  getDeleteRowRequestBody: (args: {
+    pkClause: WhereClause;
+    tableName: string;
+    schemaName: string;
+    columnInfo: BaseTableColumn[];
+    source: string;
+    tableConfiguration: TableConfig;
+  }) =>
+    | {
+        type: string;
+        args: {
+          source: string;
+          table: {
+            name: string;
+            schema: string;
+          };
+          where: WhereClause;
+        };
+      }
+    | {
+        query: string;
+        variables: null;
+      };
+  processDeleteRowData: (data: Record<string, any>) => number;
+};
+
+export type GenerateBulkDeleteRowRequest = {
+  endpoint: string;
+  getBulkDeleteRowRequestBody: (args: {
+    pkClauses: WhereClause[];
+    tableName: string;
+    schemaName: string;
+    columnInfo: BaseTableColumn[];
+    source: string;
+    tableConfiguration: TableConfig;
+  }) =>
+    | {
+        type: string;
+        source: string;
+        args: {
+          type: string;
+          args: {
+            source: string;
+            table: {
+              name: string;
+              schema: string;
+            };
+            where: WhereClause;
+          };
+        }[];
+      }
+    | {
+        query: string;
+        variables: null;
+      };
+  processBulkDeleteRowData: (data: Record<string, any>) => number;
+};
 export type ViolationActions =
  | 'restrict'
  | 'no action'
--- a/console/src/metadata/queryUtils.ts
+++ b/console/src/metadata/queryUtils.ts
@ -126,6 +126,9 @@ export const getMetadataQuery = (
    case 'bigquery':
      prefix = 'bigquery_';
      break;
+    case 'citus':
+      prefix = 'citus_';
+      break;
    case 'postgres':
    default:
      prefix = 'pg_';
--- a/console/src/metadata/sourcesUtils.ts
+++ b/console/src/metadata/sourcesUtils.ts
@ -1,4 +1,4 @@
-import { Driver } from '../dataSources';
+import { Driver, sourceNames } from '../dataSources';
 import {
  ConnectionPoolSettings,
  IsolationLevelOptions,
@ -67,7 +67,7 @@ export const addSource = (
  }

  return {
-    type: 'pg_add_source',
+    type: `${driver === 'postgres' ? 'pg' : 'citus'}_add_source`,
    args: {
      name: payload.name,
      configuration: {
@ -88,15 +88,15 @@ export const addSource = (
 export const removeSource = (driver: Driver, name: string) => {
  let prefix = '';
  switch (driver) {
-    case 'mssql':
+    case sourceNames.mssql:
      prefix = 'mssql_';
      break;
-    case 'mysql':
-      prefix = 'mysql_';
-      break;
-    case 'bigquery':
+    case sourceNames.bigquery:
      prefix = 'bigquery_';
      break;
+    case sourceNames.citus:
+      prefix = 'citus_';
+      break;
    default:
      prefix = 'pg_';
  }
--- a/console/src/metadata/types.ts
+++ b/console/src/metadata/types.ts
@ -941,7 +941,7 @@ export interface HasuraMetadataV2 {

 export interface MetadataDataSource {
  name: string;
-  kind?: 'postgres' | 'mysql' | 'mssql' | 'bigquery';
+  kind?: 'postgres' | 'mysql' | 'mssql' | 'bigquery' | 'citus';
  configuration?: {
    connection_info?: SourceConnectionInfo;
    // pro-only feature