diff --git a/src/executor/select.rs b/src/executor/select.rs
index 73815f1e2..405a7c3d9 100644
--- a/src/executor/select.rs
+++ b/src/executor/select.rs
@@ -1,7 +1,7 @@
 use crate::{
     error::*, ConnectionTrait, DbBackend, EntityTrait, FromQueryResult, IdenStatic, Iterable,
     ModelTrait, PartialModelTrait, PrimaryKeyToColumn, QueryResult, QuerySelect, Select, SelectA,
-    SelectB, SelectTwo, SelectTwoMany, Statement, StreamTrait, TryGetableMany,
+    SelectB, SelectBoth, SelectTwo, SelectTwoMany, Statement, StreamTrait, TryGetableMany,
 };
 use futures::{Stream, TryStreamExt};
 use sea_query::SelectStatement;
@@ -70,7 +70,7 @@ where
     model: PhantomData<M>,
 }
 
-/// Defines a type to get two Models
+/// Defines a type to get two Models, with the second being optional
 #[derive(Clone, Debug)]
 pub struct SelectTwoModel<M, N>
 where
@@ -80,6 +80,16 @@ where
     model: PhantomData<(M, N)>,
 }
 
+/// Defines a type to get two Models
+#[derive(Clone, Debug)]
+pub struct SelectBothModel<M, N>
+where
+    M: FromQueryResult,
+    N: FromQueryResult,
+{
+    model: PhantomData<(M, N)>,
+}
+
 impl<T, C> SelectorTrait for SelectGetableValue<T, C>
 where
     T: TryGetableMany,
@@ -130,6 +140,21 @@ where
     }
 }
 
+impl<M, N> SelectorTrait for SelectBothModel<M, N>
+where
+    M: FromQueryResult + Sized,
+    N: FromQueryResult + Sized,
+{
+    type Item = (M, N);
+
+    fn from_raw_query_result(res: QueryResult) -> Result<Self::Item, DbErr> {
+        Ok((
+            M::from_query_result(&res, SelectA.as_str())?,
+            N::from_query_result(&res, SelectB.as_str())?,
+        ))
+    }
+}
+
 impl<E> Select<E>
 where
     E: EntityTrait,
@@ -617,6 +642,85 @@ where
     // we should only count the number of items of the parent model
 }
 
+impl<E, F> SelectBoth<E, F>
+where
+    E: EntityTrait,
+    F: EntityTrait,
+{
+    /// Perform a conversion into a [SelectBothModel]
+    pub fn into_model<M, N>(self) -> Selector<SelectBothModel<M, N>>
+    where
+        M: FromQueryResult,
+        N: FromQueryResult,
+    {
+        Selector {
+            query: self.query,
+            selector: SelectBothModel { model: PhantomData },
+        }
+    }
+
+    /// Perform a conversion into a [SelectBothModel] with [PartialModel](PartialModelTrait)
+    pub fn into_partial_model<M, N>(self) -> Selector<SelectBothModel<M, N>>
+    where
+        M: PartialModelTrait,
+        N: PartialModelTrait,
+    {
+        let select = QuerySelect::select_only(self);
+        let select = M::select_cols(select);
+        let select = N::select_cols(select);
+        select.into_model::<M, N>()
+    }
+
+    /// Convert the Models into JsonValue
+    #[cfg(feature = "with-json")]
+    pub fn into_json(self) -> Selector<SelectBothModel<JsonValue, JsonValue>> {
+        Selector {
+            query: self.query,
+            selector: SelectBothModel { model: PhantomData },
+        }
+    }
+
+    /// Get one Model from the Select query
+    pub async fn one<'a, C>(self, db: &C) -> Result<Option<(E::Model, F::Model)>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        self.into_model().one(db).await
+    }
+
+    /// Get all Models from the Select query
+    pub async fn all<'a, C>(self, db: &C) -> Result<Vec<(E::Model, F::Model)>, DbErr>
+    where
+        C: ConnectionTrait,
+    {
+        self.into_model().all(db).await
+    }
+
+    /// Stream the results of a Select operation on a Model
+    pub async fn stream<'a: 'b, 'b, C>(
+        self,
+        db: &'a C,
+    ) -> Result<impl Stream<Item = Result<(E::Model, F::Model), DbErr>> + 'b, DbErr>
+    where
+        C: ConnectionTrait + StreamTrait + Send,
+    {
+        self.into_model().stream(db).await
+    }
+
+    /// Stream the result of the operation with PartialModel
+    pub async fn stream_partial_model<'a: 'b, 'b, C, M, N>(
+        self,
+        db: &'a C,
+    ) -> Result<impl Stream<Item = Result<(M, N), DbErr>> + 'b + Send, DbErr>
+    where
+        C: ConnectionTrait + StreamTrait + Send,
+        M: PartialModelTrait + Send + 'b,
+        N: PartialModelTrait + Send + 'b,
+    {
+        self.into_partial_model().stream(db).await
+    }
+}
+
 impl<S> Selector<S>
 where
     S: SelectorTrait,
diff --git a/src/query/combine.rs b/src/query/combine.rs
index 9ffb8e6ed..75e3dabc6 100644
--- a/src/query/combine.rs
+++ b/src/query/combine.rs
@@ -1,5 +1,6 @@
 use crate::{
-    ColumnTrait, EntityTrait, IdenStatic, Iterable, QueryTrait, Select, SelectTwo, SelectTwoMany,
+    ColumnTrait, EntityTrait, IdenStatic, Iterable, QueryTrait, Select, SelectBoth, SelectTwo,
+    SelectTwoMany,
 };
 use core::marker::PhantomData;
 pub use sea_query::JoinType;
@@ -72,7 +73,7 @@ where
         self
     }
 
-    /// Selects and Entity and returns it together with the Entity from `Self`
+    /// Selects an optional Entity and returns it together with the Entity from `Self`
     pub fn select_also<F>(mut self, _: F) -> SelectTwo<E, F>
     where
         F: EntityTrait,
@@ -89,6 +90,15 @@ where
         self = self.apply_alias(SelectA.as_str());
         SelectTwoMany::new(self.into_query())
     }
+
+    /// Selects an Entity and returns it together with the Entity from `Self`
+    pub fn select_both<F>(mut self, _: F) -> SelectBoth<E, F>
+    where
+        F: EntityTrait,
+    {
+        self = self.apply_alias(SelectA.as_str());
+        SelectBoth::new(self.into_query())
+    }
 }
 
 impl<E, F> SelectTwo<E, F>
@@ -140,6 +150,28 @@ where
     }
 }
 
+impl<E, F> SelectBoth<E, F>
+where
+    E: EntityTrait,
+    F: EntityTrait,
+{
+    pub(crate) fn new(query: SelectStatement) -> Self {
+        Self::new_without_prepare(query).prepare_select()
+    }
+
+    pub(crate) fn new_without_prepare(query: SelectStatement) -> Self {
+        Self {
+            query,
+            entity: PhantomData,
+        }
+    }
+
+    fn prepare_select(mut self) -> Self {
+        prepare_select_two::<F, Self>(&mut self);
+        self
+    }
+}
+
 fn prepare_select_two<F, S>(selector: &mut S)
 where
     F: EntityTrait,
diff --git a/src/query/join.rs b/src/query/join.rs
index 09f2e3857..263fd89c4 100644
--- a/src/query/join.rs
+++ b/src/query/join.rs
@@ -1,6 +1,6 @@
 use crate::{
     join_tbl_on_condition, unpack_table_ref, ColumnTrait, EntityTrait, IdenStatic, Iterable,
-    Linked, QuerySelect, Related, Select, SelectA, SelectB, SelectTwo, SelectTwoMany,
+    Linked, QuerySelect, Related, Select, SelectA, SelectB, SelectBoth, SelectTwo, SelectTwoMany,
 };
 pub use sea_query::JoinType;
 use sea_query::{Alias, Condition, Expr, IntoIden, SeaRc, SelectExpr};
@@ -53,6 +53,15 @@ where
         self.left_join(r).select_also(r)
     }
 
+    /// Inner Join with a Related Entity and select both Entity.
+    pub fn find_both_related<R>(self, r: R) -> SelectBoth<E, R>
+    where
+        R: EntityTrait,
+        E: Related<R>,
+    {
+        self.inner_join(r).select_both(r)
+    }
+
     /// Left Join with a Related Entity and select the related Entity as a `Vec`
     pub fn find_with_related<R>(self, r: R) -> SelectTwoMany<E, R>
     where
diff --git a/src/query/select.rs b/src/query/select.rs
index f747fb582..cea1bed2e 100644
--- a/src/query/select.rs
+++ b/src/query/select.rs
@@ -14,7 +14,7 @@ where
     pub(crate) entity: PhantomData<E>,
 }
 
-/// Defines a structure to perform a SELECT operation on two Models
+/// Defines a structure to perform a SELECT operation on two Models, with the second Model being optional
 #[derive(Clone, Debug)]
 pub struct SelectTwo<E, F>
 where
@@ -36,6 +36,17 @@ where
     pub(crate) entity: PhantomData<(E, F)>,
 }
 
+/// Defines a structure to perform a SELECT operation on two Models
+#[derive(Clone, Debug)]
+pub struct SelectBoth<E, F>
+where
+    E: EntityTrait,
+    F: EntityTrait,
+{
+    pub(crate) query: SelectStatement,
+    pub(crate) entity: PhantomData<(E, F)>,
+}
+
 /// Performs a conversion to [SimpleExpr]
 pub trait IntoSimpleExpr {
     /// Method to perform the conversion
@@ -78,6 +89,18 @@ macro_rules! impl_trait {
                 &mut self.query
             }
         }
+
+        impl<E, F> $trait for SelectBoth<E, F>
+        where
+            E: EntityTrait,
+            F: EntityTrait,
+        {
+            type QueryStatement = SelectStatement;
+
+            fn query(&mut self) -> &mut SelectStatement {
+                &mut self.query
+            }
+        }
     };
 }
 
@@ -175,3 +198,4 @@ macro_rules! select_two {
 
 select_two!(SelectTwo);
 select_two!(SelectTwoMany);
+select_two!(SelectBoth);
diff --git a/tests/relational_tests.rs b/tests/relational_tests.rs
index 499275342..1b51c8614 100644
--- a/tests/relational_tests.rs
+++ b/tests/relational_tests.rs
@@ -289,6 +289,84 @@ pub async fn inner_join() {
     ctx.delete().await;
 }
 
+#[sea_orm_macros::test]
+#[cfg(any(
+    feature = "sqlx-mysql",
+    feature = "sqlx-sqlite",
+    feature = "sqlx-postgres"
+))]
+pub async fn find_both_related() {
+    let ctx = TestContext::new("test_find_both_related").await;
+    create_tables(&ctx.db).await.unwrap();
+
+    let bakery = bakery::ActiveModel {
+        name: Set("SeaSide Bakery".to_owned()),
+        profit_margin: Set(10.4),
+        ..Default::default()
+    }
+    .insert(&ctx.db)
+    .await
+    .expect("could not insert bakery");
+
+    let customer_kate = customer::ActiveModel {
+        name: Set("Kate".to_owned()),
+        ..Default::default()
+    }
+    .insert(&ctx.db)
+    .await
+    .expect("could not insert customer");
+
+    let _customer_jim = customer::ActiveModel {
+        name: Set("Jim".to_owned()),
+        ..Default::default()
+    }
+    .insert(&ctx.db)
+    .await
+    .expect("could not insert customer");
+
+    let kate_order_1 = order::ActiveModel {
+        bakery_id: Set(bakery.id),
+        customer_id: Set(customer_kate.id),
+        total: Set(dec!(15.10)),
+        placed_at: Set(Utc::now().naive_utc()),
+
+        ..Default::default()
+    }
+    .insert(&ctx.db)
+    .await
+    .expect("could not insert order");
+
+    let kate_order_2 = order::ActiveModel {
+        bakery_id: Set(bakery.id),
+        customer_id: Set(customer_kate.id),
+        total: Set(dec!(100.00)),
+        placed_at: Set(Utc::now().naive_utc()),
+
+        ..Default::default()
+    }
+    .insert(&ctx.db)
+    .await
+    .expect("could not insert order");
+
+    let results: Vec<(order::Model, customer::Model)> = order::Entity::find()
+        .find_both_related(customer::Entity)
+        .all(&ctx.db)
+        .await
+        .unwrap();
+
+    assert_eq!(results.len(), 2);
+    assert!(results
+        .iter()
+        .any(|result| result.1.name == customer_kate.name.clone()
+            && result.0.total == kate_order_1.total));
+    assert!(results
+        .iter()
+        .any(|result| result.1.name == customer_kate.name.clone()
+            && result.0.total == kate_order_2.total));
+
+    ctx.delete().await;
+}
+
 #[sea_orm_macros::test]
 #[cfg(any(
     feature = "sqlx-mysql",