diff --git a/lib/arke_postgres.ex b/lib/arke_postgres.ex
index 0e8499d..da93a11 100644
--- a/lib/arke_postgres.ex
+++ b/lib/arke_postgres.ex
@@ -20,8 +20,8 @@ defmodule ArkePostgres do
     case check_env() do
       {:ok, nil} ->
         try do
+          projects = Query.get_project_record()
 
-          projects =Query.get_project_record()
           Enum.each(projects, fn %{id: project_id} = _project ->
             start_managers(project_id)
           end)
@@ -31,10 +31,16 @@ defmodule ArkePostgres do
           _ in DBConnection.ConnectionError ->
             IO.inspect("ConnectionError")
             :error
+
           err in Postgrex.Error ->
-            %{message: message,postgres: %{code: code, message: postgres_message}} = err
-            parsed_message = %{context: "postgrex_error", message: "#{message || postgres_message}"}
-            IO.inspect(parsed_message,syntax_colors: [string: :red,atom: :cyan, ])
+            %{message: message, postgres: %{code: code, message: postgres_message}} = err
+
+            parsed_message = %{
+              context: "postgrex_error",
+              message: "#{message || postgres_message}"
+            }
+
+            IO.inspect(parsed_message, syntax_colors: [string: :red, atom: :cyan])
             :error
         end
 
@@ -67,64 +73,82 @@ defmodule ArkePostgres do
     end
   end
 
-  defp start_managers(project_id) when is_binary(project_id), do: start_managers(String.to_atom(project_id))
+  defp start_managers(project_id) when is_binary(project_id),
+    do: start_managers(String.to_atom(project_id))
+
   defp start_managers(project_id) do
     {parameters, arke_list, groups} = Query.get_manager_units(project_id)
 
-    Arke.handle_manager(parameters,project_id,:parameter)
-    Arke.handle_manager(arke_list,project_id,:arke)
-    Arke.handle_manager(groups,project_id,:group)
-
+    Arke.handle_manager(parameters, project_id, :parameter)
+    Arke.handle_manager(arke_list, project_id, :arke)
+    Arke.handle_manager(groups, project_id, :group)
   end
 
-  def create(project, %{arke_id: arke_id} = unit) do
+  def create(project, unit, opts \\ [])
+
+  def create(project, %{arke_id: arke_id} = unit, opts),
+    do: create(project, [unit], opts)
+
+  def create(_project, [], _opts), do: {:ok, 0, [], []}
+
+  def create(project, [%{arke_id: arke_id} | _] = unit_list, opts) do
     arke = Arke.Boundary.ArkeManager.get(arke_id, project)
-    case handle_create(project, arke, unit) do
+
+    case handle_create(project, arke, unit_list, opts) do
       {:ok, unit} ->
         {:ok,
          Arke.Core.Unit.update(unit, metadata: Map.merge(unit.metadata, %{project: project}))}
 
+      {:ok, count, valid, errors} ->
+        {:ok, count,
+         Enum.map(valid, fn unit ->
+           Arke.Core.Unit.update(unit, metadata: Map.merge(unit.metadata, %{project: project}))
+         end), errors}
+
       {:error, errors} ->
-        {:error, handle_changeset_errros(errors)}
+        {:error, errors}
     end
   end
 
   defp handle_create(
          project,
          %{data: %{type: "table"}} = arke,
-         %{data: data, metadata: metadata} = unit
+         [%{data: data, metadata: metadata} = unit | _] = _,
+         _opts
        ) do
+    # todo: handle bulk?
     # todo: remove once the project is not needed anymore
     data = data |> Map.merge(%{metadata: Map.delete(metadata, :project)}) |> data_as_klist
     Table.insert(project, arke, data)
     {:ok, unit}
   end
 
-  defp handle_create(project, %{data: %{type: "arke"}} = arke, unit) do
-    case ArkeUnit.insert(project, arke, unit) do
-      {:ok, %{id: id, inserted_at: inserted_at, updated_at: updated_at}} ->
-        {:ok,
-         Arke.Core.Unit.update(unit, id: id, inserted_at: inserted_at, updated_at: updated_at)}
-
-      {:error, errors} ->
-        {:error, errors}
-    end
-  end
+  defp handle_create(project, %{data: %{type: "arke"}} = arke, unit_list, opts),
+    do: ArkeUnit.insert(project, arke, unit_list, opts)
 
-  defp handle_create(proj, arke, unit) do
+  defp handle_create(_project, _arke, _unit, _opts) do
     {:error, "arke type not supported"}
   end
 
-  def update(project, %{arke_id: arke_id} = unit) do
+  def update(project, unit, opts \\ [])
+
+  def update(project, %{arke_id: arke_id} = unit, opts),
+    do: update(project, [unit], opts)
+
+  def update(_project, [], _opts), do: {:ok, 0, [], []}
+
+  def update(project, [%{arke_id: arke_id} | _] = unit_list, opts) do
     arke = Arke.Boundary.ArkeManager.get(arke_id, project)
-    {:ok, unit} = handle_update(project, arke, unit)
+    handle_update(project, arke, unit_list, opts)
   end
 
   def handle_update(
         project,
         %{data: %{type: "table"}} = arke,
-        %{data: data, metadata: metadata} = unit
+        [%{data: data, metadata: metadata} = unit | _] = _,
+        _opts
       ) do
+    # todo: handle bulk?
     data =
       unit
       |> filter_primary_keys(false)
@@ -138,21 +162,30 @@ defmodule ArkePostgres do
     {:ok, unit}
   end
 
-  def handle_update(project, %{data: %{type: "arke"}} = arke, unit) do
-    ArkeUnit.update(project, arke, unit)
-    {:ok, unit}
-  end
+  def handle_update(project, %{data: %{type: "arke"}} = arke, unit_list, opts),
+    do: ArkeUnit.update(project, arke, unit_list, opts)
 
-  def handle_update(_, _, _) do
+  def handle_update(_project, _arke, _unit, _opts) do
     {:error, "arke type not supported"}
   end
 
-  def delete(project, %{arke_id: arke_id} = unit) do
+  def delete(project, unit, opts \\ [])
+
+  def delete(project, %{arke_id: arke_id} = unit, opts), do: delete(project, [unit], opts)
+
+  def delete(project, [], opts), do: {:ok, nil}
+
+  def delete(project, [%{arke_id: arke_id} | _] = unit_list, opts) do
     arke = Arke.Boundary.ArkeManager.get(arke_id, project)
-    handle_delete(project, arke, unit)
+    handle_delete(project, arke, unit_list)
   end
 
-  defp handle_delete(project, %{data: %{type: "table"}} = arke, %{metadata: metadata} = unit) do
+  defp handle_delete(
+         project,
+         %{data: %{type: "table"}} = arke,
+         [%{metadata: metadata} = unit | _] = _
+       ) do
+    # todo: handle bulk?
     metadata = Map.delete(metadata, :project)
 
     where = unit |> filter_primary_keys(true) |> Map.put_new(:metadata, metadata) |> data_as_klist
@@ -163,11 +196,8 @@ defmodule ArkePostgres do
     end
   end
 
-  defp handle_delete(project, %{data: %{type: "arke"}} = arke, unit) do
-    case ArkeUnit.delete(project, arke, unit) do
-      {:ok, _} -> {:ok, nil}
-      {:error, msg} -> {:error, msg}
-    end
+  defp handle_delete(project, %{data: %{type: "arke"}} = arke, unit_list) do
+    ArkeUnit.delete(project, arke, unit_list)
   end
 
   defp handle_delete(_, _, _) do
@@ -202,13 +232,6 @@ defmodule ArkePostgres do
     Enum.to_list(data)
   end
 
-  defp handle_changeset_errros(errors)when is_binary(errors), do: errors
-  defp handle_changeset_errros(errors) do
-    Enum.map(errors, fn {field, detail} ->
-      "#{field}: #{render_detail(detail)}"
-    end)
-  end
-
   defp render_detail({message, values}) do
     Enum.reduce(values, message, fn {k, v}, acc ->
       String.replace(acc, "%{#{k}}", to_string(v))
diff --git a/lib/arke_postgres/arke_unit.ex b/lib/arke_postgres/arke_unit.ex
index 1506055..61b678e 100644
--- a/lib/arke_postgres/arke_unit.ex
+++ b/lib/arke_postgres/arke_unit.ex
@@ -19,24 +19,54 @@ defmodule ArkePostgres.ArkeUnit do
 
   @record_fields [:id, :data, :metadata, :inserted_at, :updated_at]
 
-  def insert(project, arke, %{data: data} = unit) do
-    row = [
-      id: handle_id(unit.id),
-      arke_id: Atom.to_string(unit.arke_id),
-      data: encode_unit_data(arke, data),
-      metadata: unit.metadata,
-      inserted_at: unit.inserted_at,
-      updated_at: unit.updated_at
-    ]
-
-    case ArkePostgres.Repo.insert(ArkePostgres.Tables.ArkeUnit.changeset(Enum.into(row, %{})),
-           prefix: project
-         ) do
-      {:ok, record} ->
-        {:ok, record}
-
-      {:error, changeset} ->
-        {:error, changeset.errors}
+  def insert(project, arke, unit_list, opts \\ []) do
+    now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second)
+
+    %{unit_list: updated_unit_list, records: records} =
+      Enum.reduce(unit_list, %{unit_list: [], records: []}, fn unit, acc ->
+        id = handle_id(unit.id)
+
+        updated_unit =
+          unit |> Map.put(:id, id) |> Map.put(:inserted_at, now) |> Map.put(:updated_at, now)
+
+        acc
+        |> Map.put(:unit_list, [updated_unit | acc.unit_list])
+        |> Map.put(:records, [
+          %{
+            id: id,
+            arke_id: Atom.to_string(unit.arke_id),
+            data: encode_unit_data(arke, unit.data),
+            metadata: unit.metadata,
+            inserted_at: now,
+            updated_at: now
+          }
+          | acc.records
+        ])
+      end)
+
+    case(
+      ArkePostgres.Repo.insert_all(
+        ArkePostgres.Tables.ArkeUnit,
+        records,
+        prefix: project,
+        returning: true
+      )
+    ) do
+      {0, _} ->
+        {:error, Error.create(:insert, "no records inserted")}
+
+      {count, inserted} ->
+        inserted_ids = Enum.map(inserted, & &1.id)
+
+        {valid, errors} =
+          Enum.split_with(updated_unit_list, fn unit ->
+            unit.id in inserted_ids
+          end)
+
+        case opts[:bulk] do
+          true -> {:ok, count, valid, errors}
+          _ -> {:ok, List.first(valid)}
+        end
     end
   end
 
@@ -46,23 +76,51 @@ defmodule ArkePostgres.ArkeUnit do
   # TODO handle error
   defp handle_id(id), do: id
 
-  def update(project, arke, %{data: data} = unit, where \\ []) do
-    where = Keyword.put_new(where, :arke_id, Atom.to_string(unit.arke_id))
-    where = Keyword.put_new(where, :id, Atom.to_string(unit.id))
+  def update(project, arke, unit_list, opts) do
+    records =
+      Enum.map(unit_list, fn unit ->
+        %{
+          id: to_string(unit.id),
+          arke_id: to_string(arke.id),
+          data: encode_unit_data(arke, unit.data),
+          metadata: Map.delete(unit.metadata, :project),
+          inserted_at: DateTime.to_naive(unit.inserted_at) |> NaiveDateTime.truncate(:second),
+          updated_at: DateTime.to_naive(unit.updated_at)
+        }
+      end)
+
+    case ArkePostgres.Repo.insert_all(
+           ArkePostgres.Tables.ArkeUnit,
+           records,
+           prefix: project,
+           on_conflict: {:replace_all_except, [:id]},
+           conflict_target: :id,
+           returning: true
+         ) do
+      {0, _} ->
+        {:error, Error.create(:update, "no records updated")}
 
-    row = [
-      data: encode_unit_data(arke, data),
-      metadata: Map.delete(unit.metadata, :project),
-      updated_at: unit.updated_at
-    ]
+      {count, updated} ->
+        updated_ids = Enum.map(updated, & &1.id)
 
-    query = from("arke_unit", where: ^where, update: [set: ^row])
-    ArkePostgres.Repo.update_all(query, [], prefix: project)
+        {valid, errors} =
+          Enum.split_with(unit_list, fn unit ->
+            to_string(unit.id) in updated_ids
+          end)
+
+        case opts[:bulk] do
+          true -> {:ok, count, valid, errors}
+          _ -> {:ok, List.first(valid)}
+        end
+    end
   end
 
-  def delete(project, arke, unit) do
-    where = [arke_id: Atom.to_string(arke.id), id: Atom.to_string(unit.id)]
-    query = from(a in "arke_unit", where: ^where)
+  def delete(project, arke, unit_list) do
+    query =
+      from(a in "arke_unit",
+        where: a.arke_id == ^Atom.to_string(arke.id),
+        where: a.id in ^Enum.map(unit_list, &Atom.to_string(&1.id))
+      )
 
     case ArkePostgres.Repo.delete_all(query, prefix: project) do
       {0, nil} ->
diff --git a/lib/arke_postgres/query.ex b/lib/arke_postgres/query.ex
index 4d867f9..067c25d 100644
--- a/lib/arke_postgres/query.ex
+++ b/lib/arke_postgres/query.ex
@@ -23,6 +23,7 @@ defmodule ArkePostgres.Query do
         action
       ) do
     base_query(arke_query, action)
+    |> handle_paths_join(filters, orders)
     |> handle_filters(filters)
     |> handle_orders(orders)
     |> handle_offset(offset)
@@ -49,13 +50,16 @@ defmodule ArkePostgres.Query do
 
   def execute(query, :pseudo_query), do: generate_query(query, :pseudo_query)
 
-  def get_column(%{data: %{persistence: "arke_parameter"}} = parameter),
-    do: get_arke_column(parameter)
+  def get_column(column), do: get_column(column, false)
 
-  def get_column(%{data: %{persistence: "table_column"}} = parameter),
+  def get_column(%{data: %{persistence: "arke_parameter"}} = parameter, joined),
+    do: get_arke_column(parameter, joined)
+
+  def get_column(%{data: %{persistence: "table_column"}} = parameter, _joined),
     do: get_table_column(parameter)
 
   def remove_arke_system(metadata, project_id) when project_id == :arke_system, do: metadata
+
   def remove_arke_system(metadata, project_id) do
     case Map.get(metadata, "project") do
       "arke_system" -> Map.delete(metadata, "project")
@@ -72,7 +76,10 @@ defmodule ArkePostgres.Query do
   end
 
   def get_manager_units(project_id) do
-    arke_link =%{id: :arke_link, data: %{parameters: [%{id: :type},%{id: :child_id},%{id: :parent_id},%{id: :metadata}]}}
+    arke_link = %{
+      id: :arke_link,
+      data: %{parameters: [%{id: :type}, %{id: :child_id}, %{id: :parent_id}, %{id: :metadata}]}
+    }
 
     links =
       from(q in table_query(arke_link, nil), where: q.type in ["parameter", "group"])
@@ -81,7 +88,6 @@ defmodule ArkePostgres.Query do
     parameter_links = Enum.filter(links, fn x -> x.type == "parameter" end)
     group_links = Enum.filter(links, fn x -> x.type == "group" end)
 
-
     parameters_id = Arke.Utils.DefaultData.get_parameters_id()
 
     list_arke_id = Arke.Utils.DefaultData.get_arke_id()
@@ -93,13 +99,15 @@ defmodule ArkePostgres.Query do
     units_map = Map.new(unit_list, &{&1.id, &1.metadata})
     parameter_links = merge_unit_metadata(parameter_links, units_map, project_id)
 
-    parameters = parse_parameters(Enum.filter(unit_list, fn u -> u.arke_id in parameters_id end), project_id)
+    parameters =
+      parse_parameters(Enum.filter(unit_list, fn u -> u.arke_id in parameters_id end), project_id)
 
     arke_list =
       parse_arke_list(
         Enum.filter(unit_list, fn u -> u.arke_id == "arke" end),
         parameter_links
       )
+
     groups =
       parse_groups(
         Enum.filter(unit_list, fn u -> u.arke_id == "group" end),
@@ -123,27 +131,45 @@ defmodule ArkePostgres.Query do
           Enum.filter(parameter_links, fn x -> x.parent_id == id end),
           [],
           fn p, new_params ->
-            [%{id: String.to_atom(p.child_id), metadata: Enum.reduce(p.metadata,%{}, fn {key, val}, acc -> Map.put(acc, String.to_atom(key), val) end)} | new_params]
+            [
+              %{
+                id: String.to_atom(p.child_id),
+                metadata:
+                  Enum.reduce(p.metadata, %{}, fn {key, val}, acc ->
+                    Map.put(acc, String.to_atom(key), val)
+                  end)
+              }
+              | new_params
+            ]
           end
         )
-        updated_data = Enum.reduce(unit.data,%{}, fn {k,db_data},acc -> Map.put(acc,String.to_atom(k),db_data["value"]) end)
+
+      updated_data =
+        Enum.reduce(unit.data, %{}, fn {k, db_data}, acc ->
+          Map.put(acc, String.to_atom(k), db_data["value"])
+        end)
         |> Map.put(:id, id)
         |> Map.put(:metadata, metadata)
-        |> Map.update(:parameters,[], fn current -> params ++ current end)
+        |> Map.update(:parameters, [], fn current -> params ++ current end)
 
-      [ updated_data | new_arke_list]
+      [updated_data | new_arke_list]
     end)
   end
 
-  defp parse_parameters(parameter_list, project_id)do
-    Enum.reduce(parameter_list, [], fn %{id: id, arke_id: arke_id, metadata: metadata} = unit, new_parameter_list ->
+  defp parse_parameters(parameter_list, project_id) do
+    Enum.reduce(parameter_list, [], fn %{id: id, arke_id: arke_id, metadata: metadata} = unit,
+                                       new_parameter_list ->
       parsed_metadata = remove_arke_system(metadata, project_id)
 
-      updated_data = Enum.reduce(unit.data,%{}, fn {k,db_data},acc -> Map.put(acc,String.to_atom(k),db_data["value"]) end)
-                     |> Map.put(:id, id)
-                     |> Map.put(:type, arke_id)
-                     |> Map.put(:metadata, parsed_metadata)
-      [ updated_data | new_parameter_list]
+      updated_data =
+        Enum.reduce(unit.data, %{}, fn {k, db_data}, acc ->
+          Map.put(acc, String.to_atom(k), db_data["value"])
+        end)
+        |> Map.put(:id, id)
+        |> Map.put(:type, arke_id)
+        |> Map.put(:metadata, parsed_metadata)
+
+      [updated_data | new_parameter_list]
     end)
   end
 
@@ -157,23 +183,31 @@ defmodule ArkePostgres.Query do
             [%{id: String.to_atom(p.child_id), metadata: p.metadata} | new_params]
           end
         )
-      updated_data = Enum.reduce(unit.data,%{}, fn {k,db_data},acc -> Map.put(acc,String.to_atom(k),db_data["value"]) end)
-                     |> Map.put(:id, id)
-                     |> Map.put(:metadata, metadata)
-                     |> Map.update(:arke_list,[], fn db_arke_list ->
-      Enum.reduce(db_arke_list,[], fn key,acc ->
-        case Enum.find(arke_list, fn %{id: id, metadata: _metadata} -> to_string(id) == key end) do
-          nil -> [key|acc]
-          data ->
-            [data|acc]
-        end
-       end)
-      end)
-      [ updated_data | new_groups]
+
+      updated_data =
+        Enum.reduce(unit.data, %{}, fn {k, db_data}, acc ->
+          Map.put(acc, String.to_atom(k), db_data["value"])
+        end)
+        |> Map.put(:id, id)
+        |> Map.put(:metadata, metadata)
+        |> Map.update(:arke_list, [], fn db_arke_list ->
+          Enum.reduce(db_arke_list, [], fn key, acc ->
+            case Enum.find(arke_list, fn %{id: id, metadata: _metadata} ->
+                   to_string(id) == key
+                 end) do
+              nil ->
+                [key | acc]
+
+              data ->
+                [data | acc]
+            end
+          end)
+        end)
+
+      [updated_data | new_groups]
     end)
   end
 
-
   ######################################################################################################################
   # PRIVATE FUNCTIONS ##################################################################################################
   ######################################################################################################################
@@ -182,17 +216,31 @@ defmodule ArkePostgres.Query do
 
   defp base_query(%{link: nil} = _arke_query, action), do: arke_query(action)
 
-  defp base_query(%{link: %{unit: %{id: link_id},depth: depth, direction: direction,type: type}, project: project} = _arke_query, action),
-    do:
-      get_nodes(
-        project,
-        action,
-        [to_string(link_id)],
-        depth,
-        direction,
-        type
-      )
-  defp base_query(%{link: %{unit: unit_list,depth: depth, direction: direction,type: type}, project: project} = _arke_query, action) when is_list(unit_list) do
+  defp base_query(
+         %{
+           link: %{unit: %{id: link_id}, depth: depth, direction: direction, type: type},
+           project: project
+         } = _arke_query,
+         action
+       ),
+       do:
+         get_nodes(
+           project,
+           action,
+           [to_string(link_id)],
+           depth,
+           direction,
+           type
+         )
+
+  defp base_query(
+         %{
+           link: %{unit: unit_list, depth: depth, direction: direction, type: type},
+           project: project
+         } = _arke_query,
+         action
+       )
+       when is_list(unit_list) do
     get_nodes(
       project,
       action,
@@ -261,37 +309,80 @@ defmodule ArkePostgres.Query do
     Arke.Core.Unit.load(arke, record)
   end
 
+  defp handle_paths_join(query, filters, orders) do
+    paths = extract_paths(filters) ++ extract_paths(orders)
+
+    case paths do
+      [] ->
+        query
+
+      _ ->
+        conditions =
+          Enum.reduce(paths, nil, fn path, acc ->
+            condition = dynamic([q, j], ^get_column(List.first(path)) == j.id)
+            if is_nil(acc), do: condition, else: dynamic([q, j], ^acc or ^condition)
+          end)
+
+        from(q in query, join: j in "arke_unit", on: ^conditions)
+    end
+  end
+
+  defp extract_paths(items) do
+    items
+    |> Enum.flat_map(fn
+      %{base_filters: base_filters} -> Enum.map(base_filters, & &1.path)
+      %{path: path} -> [path]
+      _ -> []
+    end)
+    |> Enum.reject(&(is_nil(&1) or length(&1) == 0))
+    |> Enum.uniq()
+  end
+
   def handle_filters(query, filters) do
     Enum.reduce(filters, query, fn %{logic: logic, negate: negate, base_filters: base_filters},
                                    new_query ->
-      clause = handle_condition(logic, base_filters) |> handle_negate_condition(negate)
+      clause = handle_conditions(logic, base_filters) |> handle_negate_condition(negate)
       from(q in new_query, where: ^clause)
     end)
   end
 
-  defp handle_condition(logic, base_filters) do
+  defp handle_conditions(logic, base_filters) do
     Enum.reduce(base_filters, nil, fn %{
                                         parameter: parameter,
                                         operator: operator,
                                         value: value,
-                                        negate: negate
+                                        negate: negate,
+                                        path: path
                                       },
                                       clause ->
-      column = get_column(parameter)
-      value = get_value(parameter, value)
-
-      if is_nil(value) or operator == :isnull do
-        condition = get_nil_query(parameter, column) |> handle_negate_condition(negate)
-        add_condition_to_clause(condition, clause, logic)
+      if length(path) == 0 do
+        parameter_condition(clause, parameter, value, operator, negate, logic)
+        |> add_condition_to_clause(clause, logic)
       else
-        condition =
-          filter_query_by_operator(parameter, column, value, operator) |> handle_negate_condition(negate)
+        # todo enhance to get multi-level path
+        path_parameter = List.first(path)
 
-        add_condition_to_clause(condition, clause, logic)
+        if not is_nil(path_parameter) do
+          parameter_condition(clause, parameter, value, operator, negate, logic, true)
+          |> add_nested_condition_to_clause(clause, logic)
+        end
       end
     end)
   end
 
+  defp parameter_condition(clause, parameter, value, operator, negate, logic, joined \\ nil) do
+    column = get_column(parameter, joined)
+    value = get_value(parameter, value)
+
+    if is_nil(value) or operator == :isnull do
+      condition = get_nil_query(parameter, column) |> handle_negate_condition(negate)
+    else
+      condition =
+        filter_query_by_operator(parameter, column, value, operator)
+        |> handle_negate_condition(negate)
+    end
+  end
+
   defp handle_negate_condition(condition, true), do: dynamic([q], not (^condition))
   defp handle_negate_condition(condition, false), do: condition
 
@@ -299,10 +390,20 @@ defmodule ArkePostgres.Query do
   defp add_condition_to_clause(condition, clause, :and), do: dynamic([q], ^clause and ^condition)
   defp add_condition_to_clause(condition, clause, :or), do: dynamic([q], ^clause or ^condition)
 
+  defp add_nested_condition_to_clause(condition, nil, _), do: dynamic([_q, j], ^condition)
+
+  defp add_nested_condition_to_clause(condition, clause, :and),
+    do: dynamic([_q, j], ^clause and ^condition)
+
+  defp add_nested_condition_to_clause(condition, clause, :or),
+    do: dynamic([_q, j], ^clause or ^condition)
+
   defp handle_orders(query, orders) do
     order_by =
-      Enum.reduce(orders, [], fn %{parameter: parameter, direction: direction}, new_order_by ->
-        column = get_column(parameter)
+      Enum.reduce(orders, [], fn %{parameter: parameter, direction: direction, path: path},
+                                 new_order_by ->
+        joined = length(path) > 0
+        column = get_column(parameter, joined)
         [{direction, column} | new_order_by]
       end)
 
@@ -317,61 +418,152 @@ defmodule ArkePostgres.Query do
 
   defp get_table_column(%{id: id} = _parameter), do: dynamic([q], fragment("?", field(q, ^id)))
 
-  defp get_arke_column(%{id: id, data: %{multiple: true}} = _parameter),
-       do: dynamic([q], fragment("(? -> ? ->> 'value')::jsonb", field(q, :data), ^Atom.to_string(id)))
+  defp get_arke_column(%{id: id, data: %{multiple: true}} = _parameter, true),
+    do:
+      dynamic(
+        [_q, ..., j],
+        fragment("(? -> ? ->> 'value')::jsonb", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, data: %{multiple: true}} = _parameter, _joined),
+    do:
+      dynamic([q], fragment("(? -> ? ->> 'value')::jsonb", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :string} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :string} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::text", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :string} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::text", field(q, :data), ^Atom.to_string(id)))
 
+  defp get_arke_column(%{id: id, arke_id: :atom} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::text", field(j, :data), ^Atom.to_string(id))
+      )
 
-  defp get_arke_column(%{id: id, arke_id: :atom} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :atom} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::text", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :boolean} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :boolean} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::boolean", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :boolean} = _parameter, _joined),
     do:
       dynamic(
         [q],
         fragment("(? -> ? ->> 'value')::boolean", field(q, :data), ^Atom.to_string(id))
       )
 
-  defp get_arke_column(%{id: id, arke_id: :datetime} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :datetime} = _parameter, true),
+    do:
+      dynamic(
+        [_q, ..., j],
+        fragment("(? -> ? ->> 'value')::timestamp", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :datetime} = _parameter, _joined),
     do:
       dynamic(
         [q],
         fragment("(? -> ? ->> 'value')::timestamp", field(q, :data), ^Atom.to_string(id))
       )
 
-  defp get_arke_column(%{id: id, arke_id: :date} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :date} = _parameter, true),
+    do:
+      dynamic(
+        [_q, ..., j],
+        fragment("(? -> ? ->> 'value')::date", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :date} = _parameter, _joined),
     do:
       dynamic(
         [q],
         fragment("(? -> ? ->> 'value')::date", field(q, :data), ^Atom.to_string(id))
       )
 
-  defp get_arke_column(%{id: id, arke_id: :time} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :time} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::time", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :time} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::time", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :integer} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :integer} = _parameter, true),
+    do:
+      dynamic(
+        [_q, ..., j],
+        fragment("(? -> ? ->> 'value')::integer", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :integer} = _parameter, _joined),
     do:
       dynamic(
         [q],
         fragment("(? -> ? ->> 'value')::integer", field(q, :data), ^Atom.to_string(id))
       )
 
-  defp get_arke_column(%{id: id, arke_id: :float} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :float} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::float", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :float} = _parameter, _joined),
     do:
       dynamic([q], fragment("(? -> ? ->> 'value')::float", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :dict} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :dict} = _parameter, true),
+    do:
+      dynamic(
+        [_q, ..., j],
+        fragment("(? -> ? ->> 'value')::JSON", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :dict} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::JSON", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :list} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :list} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::JSON", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :list} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::JSON", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :link} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :link} = _parameter, true),
+    do:
+      dynamic(
+        [_q, ..., j],
+        fragment("(? -> ? ->> 'value')::text", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :link} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::text", field(q, :data), ^Atom.to_string(id)))
 
-  defp get_arke_column(%{id: id, arke_id: :dynamic} = _parameter),
+  defp get_arke_column(%{id: id, arke_id: :dynamic} = _parameter, true),
+    do:
+      dynamic(
+        [_, ..., j],
+        fragment("(? -> ? ->> 'value')::text", field(j, :data), ^Atom.to_string(id))
+      )
+
+  defp get_arke_column(%{id: id, arke_id: :dynamic} = _parameter, _joined),
     do: dynamic([q], fragment("(? -> ? ->> 'value')::text", field(q, :data), ^Atom.to_string(id)))
 
   defp get_value(_parameter, value) when is_nil(value), do: value
@@ -471,8 +663,11 @@ defmodule ArkePostgres.Query do
         fragment("? IS NULL AND (data \\? ?)", ^column, ^Atom.to_string(id))
       )
 
-  defp filter_query_by_operator(%{data: %{multiple: true}}, column, value, :eq), do: dynamic([q], fragment("jsonb_exists(?, ?)", ^column, ^value))
-  defp filter_query_by_operator(parameter, column, value, :eq), do: dynamic([q], ^column == ^value)
+  defp filter_query_by_operator(%{data: %{multiple: true}}, column, value, :eq),
+    do: dynamic([q], fragment("jsonb_exists(?, ?)", ^column, ^value))
+
+  defp filter_query_by_operator(parameter, column, value, :eq),
+    do: dynamic([q], ^column == ^value)
 
   defp filter_query_by_operator(parameter, column, value, :contains),
     do: dynamic([q], like(^column, fragment("?", ^("%" <> value <> "%"))))
@@ -492,11 +687,18 @@ defmodule ArkePostgres.Query do
   defp filter_query_by_operator(parameter, column, value, :istartswith),
     do: dynamic([q], ilike(^column, fragment("?", ^(value <> "%"))))
 
-  defp filter_query_by_operator(parameter, column, value, :lte), do: dynamic([q], ^column <= ^value)
+  defp filter_query_by_operator(parameter, column, value, :lte),
+    do: dynamic([q], ^column <= ^value)
+
   defp filter_query_by_operator(parameter, column, value, :lt), do: dynamic([q], ^column < ^value)
   defp filter_query_by_operator(parameter, column, value, :gt), do: dynamic([q], ^column > ^value)
-  defp filter_query_by_operator(parameter, column, value, :gte), do: dynamic([q], ^column >= ^value)
-  defp filter_query_by_operator(parameter, column, value, :in), do: dynamic([q], ^column in ^value)
+
+  defp filter_query_by_operator(parameter, column, value, :gte),
+    do: dynamic([q], ^column >= ^value)
+
+  defp filter_query_by_operator(parameter, column, value, :in),
+    do: dynamic([q], ^column in ^value)
+
   defp filter_query_by_operator(parameter, column, value, _), do: dynamic([q], ^column == ^value)
 
   # defp filter_query_by_operator(query, key, value, "between"), do: from q in query, where: column_table(q, ^key) == ^value
@@ -532,7 +734,9 @@ defmodule ArkePostgres.Query do
     where_field = get_where_field_by_direction(direction) |> get_where_condition_by_type(type)
     get_link_query(action, project, unit_id, link_field, tree_field, depth, where_field)
   end
-  def get_nodes(project, action, unit_id, depth, direction, type), do:  get_nodes(project, action, [unit_id], depth, direction, type)
+
+  def get_nodes(project, action, unit_id, depth, direction, type),
+    do: get_nodes(project, action, [unit_id], depth, direction, type)
 
   defp get_project(project) when is_atom(project), do: Atom.to_string(project)
   defp get_project(project), do: project
@@ -578,40 +782,43 @@ defmodule ArkePostgres.Query do
   end
 
   defp get_link_query(_action, project, unit_id_list, link_field, tree_field, depth, where_field) do
-    q = from(r in from(a in "arke_unit",
-      left_join:
-        cte in fragment(
-          @raw_cte_query,
-          literal(^link_field),
-          literal(^project),
-          literal(^link_field),
-          ^unit_id_list,
-          literal(^project),
-          literal(^project),
-          literal(^project),
-          literal(^project),
-          literal(^project),
-          literal(^project),
-          literal(^link_field),
-          literal(^tree_field),
-          ^depth
-        ),
-      where: ^where_field,
-      distinct: [a.id, cte.starting_unit],
-      select: %{
-        id: a.id,
-        arke_id: a.arke_id,
-        data: a.data,
-        metadata: a.metadata,
-        inserted_at: a.inserted_at,
-        updated_at: a.updated_at,
-        depth: cte.depth,
-        link_metadata: cte.metadata,
-        link_type: cte.type,
-        starting_unit: cte.starting_unit
-      }
-    ))
-    from x in subquery(q), select: x
-  end
+    q =
+      from(
+        r in from(a in "arke_unit",
+          left_join:
+            cte in fragment(
+              @raw_cte_query,
+              literal(^link_field),
+              literal(^project),
+              literal(^link_field),
+              ^unit_id_list,
+              literal(^project),
+              literal(^project),
+              literal(^project),
+              literal(^project),
+              literal(^project),
+              literal(^project),
+              literal(^link_field),
+              literal(^tree_field),
+              ^depth
+            ),
+          where: ^where_field,
+          distinct: [a.id, cte.starting_unit],
+          select: %{
+            id: a.id,
+            arke_id: a.arke_id,
+            data: a.data,
+            metadata: a.metadata,
+            inserted_at: a.inserted_at,
+            updated_at: a.updated_at,
+            depth: cte.depth,
+            link_metadata: cte.metadata,
+            link_type: cte.type,
+            starting_unit: cte.starting_unit
+          }
+        )
+      )
 
+    from(x in subquery(q), select: x)
+  end
 end