diff --git a/packages/server/src/api/routes/tests/row.spec.ts b/packages/server/src/api/routes/tests/row.spec.ts
index 95e714fbd6..ca9ef134cb 100644
--- a/packages/server/src/api/routes/tests/row.spec.ts
+++ b/packages/server/src/api/routes/tests/row.spec.ts
@@ -9,7 +9,13 @@ import {
 import tk from "timekeeper"
 import emitter from "../../../../src/events"
 import { outputProcessing } from "../../../utilities/rowProcessor"
-import { context, InternalTable, tenancy } from "@budibase/backend-core"
+import {
+  context,
+  InternalTable,
+  tenancy,
+  withEnv as withCoreEnv,
+  setEnv as setCoreEnv,
+} from "@budibase/backend-core"
 import { quotas } from "@budibase/pro"
 import {
   AttachmentFieldMetadata,
@@ -69,6 +75,7 @@ async function waitForEvent(
 
 describe.each([
   ["internal", undefined],
+  ["sqs", undefined],
   [DatabaseName.POSTGRES, getDatasource(DatabaseName.POSTGRES)],
   [DatabaseName.MYSQL, getDatasource(DatabaseName.MYSQL)],
   [DatabaseName.SQL_SERVER, getDatasource(DatabaseName.SQL_SERVER)],
@@ -76,6 +83,8 @@ describe.each([
   [DatabaseName.ORACLE, getDatasource(DatabaseName.ORACLE)],
 ])("/rows (%s)", (providerType, dsProvider) => {
   const isInternal = dsProvider === undefined
+  const isLucene = providerType === "lucene"
+  const isSqs = providerType === "sqs"
   const isMSSQL = providerType === DatabaseName.SQL_SERVER
   const isOracle = providerType === DatabaseName.ORACLE
   const config = setup.getConfig()
@@ -83,9 +92,17 @@ describe.each([
   let table: Table
   let datasource: Datasource | undefined
   let client: Knex | undefined
+  let envCleanup: (() => void) | undefined
 
   beforeAll(async () => {
-    await config.init()
+    await withCoreEnv({ SQS_SEARCH_ENABLE: "true" }, () => config.init())
+    if (isSqs) {
+      envCleanup = setCoreEnv({
+        SQS_SEARCH_ENABLE: "true",
+        SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()],
+      })
+    }
+
     if (dsProvider) {
       const rawDatasource = await dsProvider
       datasource = await config.createDatasource({
@@ -97,6 +114,9 @@ describe.each([
 
   afterAll(async () => {
     setup.afterAll()
+    if (envCleanup) {
+      envCleanup()
+    }
   })
 
   function saveTableRequest(
@@ -346,7 +366,7 @@ describe.each([
         expect(ids).toEqual(expect.arrayContaining(sequence))
       })
 
-    isInternal &&
+    isLucene &&
       it("row values are coerced", async () => {
         const str: FieldSchema = {
           type: FieldType.STRING,
@@ -2407,6 +2427,229 @@ describe.each([
     })
   })
 
+  // Upserting isn't yet supported in MSSQL or Oracle, see:
+  //   https://github.com/knex/knex/pull/6050
+  !isMSSQL &&
+    !isOracle &&
+    describe("relationships", () => {
+      let tableId: string
+
+      let auxData: Row[] = []
+
+      beforeAll(async () => {
+        const aux2Table = await config.api.table.save(saveTableRequest())
+        const aux2Data = await config.api.row.save(aux2Table._id!, {})
+
+        const auxTable = await config.api.table.save(
+          saveTableRequest({
+            primaryDisplay: "name",
+            schema: {
+              name: {
+                name: "name",
+                type: FieldType.STRING,
+                constraints: { presence: true },
+              },
+              age: {
+                name: "age",
+                type: FieldType.NUMBER,
+                constraints: { presence: true },
+              },
+              address: {
+                name: "address",
+                type: FieldType.STRING,
+                constraints: { presence: true },
+                visible: false,
+              },
+              link: {
+                name: "link",
+                type: FieldType.LINK,
+                tableId: aux2Table._id!,
+                relationshipType: RelationshipType.MANY_TO_MANY,
+                fieldName: "fk_aux",
+                constraints: { presence: true },
+              },
+              formula: {
+                name: "formula",
+                type: FieldType.FORMULA,
+                formula: "{{ any }}",
+                constraints: { presence: true },
+              },
+            },
+          })
+        )
+        const auxTableId = auxTable._id!
+
+        for (const name of generator.unique(() => generator.name(), 10)) {
+          auxData.push(
+            await config.api.row.save(auxTableId, {
+              name,
+              age: generator.age(),
+              address: generator.address(),
+              link: [aux2Data],
+            })
+          )
+        }
+
+        const table = await config.api.table.save(
+          saveTableRequest({
+            schema: {
+              title: {
+                name: "title",
+                type: FieldType.STRING,
+                constraints: { presence: true },
+              },
+              relWithNoSchema: {
+                name: "relWithNoSchema",
+                relationshipType: RelationshipType.ONE_TO_MANY,
+                type: FieldType.LINK,
+                tableId: auxTableId,
+                fieldName: "fk_relWithNoSchema",
+                constraints: { presence: true },
+              },
+              relWithEmptySchema: {
+                name: "relWithEmptySchema",
+                relationshipType: RelationshipType.ONE_TO_MANY,
+                type: FieldType.LINK,
+                tableId: auxTableId,
+                fieldName: "fk_relWithEmptySchema",
+                constraints: { presence: true },
+                schema: {},
+              },
+              relWithFullSchema: {
+                name: "relWithFullSchema",
+                relationshipType: RelationshipType.ONE_TO_MANY,
+                type: FieldType.LINK,
+                tableId: auxTableId,
+                fieldName: "fk_relWithFullSchema",
+                constraints: { presence: true },
+                schema: Object.keys(auxTable.schema).reduce(
+                  (acc, c) => ({ ...acc, [c]: { visible: true } }),
+                  {}
+                ),
+              },
+              relWithHalfSchema: {
+                name: "relWithHalfSchema",
+                relationshipType: RelationshipType.ONE_TO_MANY,
+                type: FieldType.LINK,
+                tableId: auxTableId,
+                fieldName: "fk_relWithHalfSchema",
+                constraints: { presence: true },
+                schema: {
+                  name: { visible: true },
+                  age: { visible: false, readonly: true },
+                },
+              },
+              relWithIllegalSchema: {
+                name: "relWithIllegalSchema",
+                relationshipType: RelationshipType.ONE_TO_MANY,
+                type: FieldType.LINK,
+                tableId: auxTableId,
+                fieldName: "fk_relWithIllegalSchema",
+                constraints: { presence: true },
+                schema: {
+                  name: { visible: true },
+                  address: { visible: true },
+                  unexisting: { visible: true },
+                },
+              },
+            },
+          })
+        )
+        tableId = table._id!
+      })
+
+      it.each([
+        ["get row", (row: Row) => config.api.row.get(tableId, row._id!)],
+        [
+          "fetch",
+          async (row: Row) => {
+            const rows = await config.api.row.fetch(tableId)
+            return rows.find(r => r._id === row._id)
+          },
+        ],
+        [
+          "search",
+          async (row: Row) => {
+            const { rows } = await config.api.row.search(tableId)
+            return rows.find(r => r._id === row._id)
+          },
+        ],
+        [
+          "from view",
+          async (row: Row) => {
+            const table = await config.api.table.get(tableId)
+            const view = await config.api.viewV2.create({
+              name: generator.guid(),
+              tableId,
+              schema: Object.keys(table.schema).reduce(
+                (acc, c) => ({ ...acc, [c]: { visible: true } }),
+                {}
+              ),
+            })
+            const { rows } = await config.api.viewV2.search(view.id)
+            return rows.find(r => r._id === row._id!)
+          },
+        ],
+        ["from original saved row", (row: Row) => row],
+      ])(
+        "can retrieve rows with populated relationships (via %s)",
+        async (__, retrieveDelegate) => {
+          const otherRows = _.sampleSize(auxData, 5)
+
+          const row = await config.api.row.save(tableId, {
+            title: generator.word(),
+            relWithNoSchema: [otherRows[0]],
+            relWithEmptySchema: [otherRows[1]],
+            relWithFullSchema: [otherRows[2]],
+            relWithHalfSchema: [otherRows[3]],
+            relWithIllegalSchema: [otherRows[4]],
+          })
+
+          const retrieved = await retrieveDelegate(row)
+          expect(retrieved).toEqual(
+            expect.objectContaining({
+              title: row.title,
+              relWithNoSchema: [
+                {
+                  _id: otherRows[0]._id,
+                  primaryDisplay: otherRows[0].name,
+                },
+              ],
+              relWithEmptySchema: [
+                {
+                  _id: otherRows[1]._id,
+                  primaryDisplay: otherRows[1].name,
+                },
+              ],
+              relWithFullSchema: [
+                {
+                  _id: otherRows[2]._id,
+                  primaryDisplay: otherRows[2].name,
+                  name: otherRows[2].name,
+                  age: otherRows[2].age,
+                  id: otherRows[2].id,
+                },
+              ],
+              relWithHalfSchema: [
+                {
+                  _id: otherRows[3]._id,
+                  primaryDisplay: otherRows[3].name,
+                  name: otherRows[3].name,
+                },
+              ],
+              relWithIllegalSchema: [
+                {
+                  _id: otherRows[4]._id,
+                  primaryDisplay: otherRows[4].name,
+                  name: otherRows[4].name,
+                },
+              ],
+            })
+          )
+        }
+      )
+    })
+
   describe("Formula fields", () => {
     let table: Table
     let otherTable: Table
diff --git a/packages/server/src/db/linkedRows/index.ts b/packages/server/src/db/linkedRows/index.ts
index 2da7e212b9..2394d00d0e 100644
--- a/packages/server/src/db/linkedRows/index.ts
+++ b/packages/server/src/db/linkedRows/index.ts
@@ -253,20 +253,34 @@ export async function squashLinksToPrimaryDisplay(
   // will populate this as we find them
   const linkedTables = [table]
   const isArray = Array.isArray(enriched)
-  let enrichedArray = !isArray ? [enriched] : enriched
-  for (let row of enrichedArray) {
+  const enrichedArray = !isArray ? [enriched] : enriched
+  for (const row of enrichedArray) {
     // this only fetches the table if its not already in array
     const rowTable = await getLinkedTable(row.tableId!, linkedTables)
-    for (let [column, schema] of Object.entries(rowTable?.schema || {})) {
+    const safeSchema =
+      (rowTable?.schema &&
+        (await sdk.tables.enrichRelationshipSchema(rowTable.schema))) ||
+      {}
+    for (let [column, schema] of Object.entries(safeSchema)) {
       if (schema.type !== FieldType.LINK || !Array.isArray(row[column])) {
         continue
       }
       const newLinks = []
-      for (let link of row[column]) {
+      for (const link of row[column]) {
         const linkTblId = link.tableId || getRelatedTableForField(table, column)
         const linkedTable = await getLinkedTable(linkTblId!, linkedTables)
         const obj: any = { _id: link._id }
         obj.primaryDisplay = getPrimaryDisplayValue(link, linkedTable)
+
+        const allowRelationshipSchemas = true // TODO
+        if (schema.schema && allowRelationshipSchemas) {
+          for (const relField of Object.entries(schema.schema)
+            .filter(([_, field]) => field.visible !== false)
+            .map(([fieldKey]) => fieldKey)) {
+            obj[relField] = link[relField]
+          }
+        }
+
         newLinks.push(obj)
       }
       row[column] = newLinks
diff --git a/packages/server/src/sdk/app/tables/getters.ts b/packages/server/src/sdk/app/tables/getters.ts
index f3e2e97795..4ce54adc30 100644
--- a/packages/server/src/sdk/app/tables/getters.ts
+++ b/packages/server/src/sdk/app/tables/getters.ts
@@ -163,7 +163,7 @@ export async function enrichRelationshipSchema(
 
     for (const relTableFieldName of Object.keys(relTable.schema)) {
       const relTableField = relTable.schema[relTableFieldName]
-      if (relTableField.type === FieldType.LINK) {
+      if ([FieldType.LINK, FieldType.FORMULA].includes(relTableField.type)) {
         continue
       }
 
@@ -171,9 +171,10 @@ export async function enrichRelationshipSchema(
         continue
       }
 
+      const isVisible = !!fieldSchema[relTableFieldName]?.visible
       const isReadonly = !!fieldSchema[relTableFieldName]?.readonly
       resultSchema[relTableFieldName] = {
-        visible: isReadonly,
+        visible: isVisible,
         readonly: isReadonly,
       }
     }
diff --git a/packages/server/src/utilities/rowProcessor/index.ts b/packages/server/src/utilities/rowProcessor/index.ts
index 4b2fd83882..acbff40602 100644
--- a/packages/server/src/utilities/rowProcessor/index.ts
+++ b/packages/server/src/utilities/rowProcessor/index.ts
@@ -3,6 +3,7 @@ import { fixAutoColumnSubType, processFormulas } from "./utils"
 import {
   cache,
   context,
+  db,
   HTTPError,
   objectStore,
   utils,
@@ -349,11 +350,19 @@ export async function outputProcessing<T extends Row[] | Row>(
   }
   // remove null properties to match internal API
   const isExternal = isExternalTableID(table._id!)
-  if (isExternal) {
+  if (isExternal || db.isSqsEnabledForTenant()) {
     for (const row of enriched) {
       for (const key of Object.keys(row)) {
         if (row[key] === null) {
           delete row[key]
+        } else if (row[key] && table.schema[key]?.type === FieldType.LINK) {
+          for (const link of row[key] || []) {
+            for (const linkKey of Object.keys(link)) {
+              if (link[linkKey] === null) {
+                delete link[linkKey]
+              }
+            }
+          }
         }
       }
     }
diff --git a/packages/server/src/utilities/rowProcessor/tests/outputProcessing.spec.ts b/packages/server/src/utilities/rowProcessor/tests/outputProcessing.spec.ts
index fd48f46016..0d0049a36e 100644
--- a/packages/server/src/utilities/rowProcessor/tests/outputProcessing.spec.ts
+++ b/packages/server/src/utilities/rowProcessor/tests/outputProcessing.spec.ts
@@ -10,6 +10,14 @@ import { outputProcessing } from ".."
 import { generator, structures } from "@budibase/backend-core/tests"
 import * as bbReferenceProcessor from "../bbReferenceProcessor"
 
+jest.mock("@budibase/backend-core", () => ({
+  ...jest.requireActual("@budibase/backend-core"),
+  db: {
+    ...jest.requireActual("@budibase/backend-core").db,
+    isSqsEnabledForTenant: () => true,
+  },
+}))
+
 jest.mock("../bbReferenceProcessor", (): typeof bbReferenceProcessor => ({
   processInputBBReference: jest.fn(),
   processInputBBReferences: jest.fn(),