perf: bq clustering (#2489)

* perf: partition by hour by default

* perf: add in BQ clustering customization

* chore: revert hour partitioning

Author: Ziinc

lib/logflare/google/bigquery/bigquery.ex

@@ -135,8 +135,12 @@ defmodule Logflare.Google.BigQuery do
     table_name = GenUtils.format_table_name(source_id)
     dataset_id = dataset_id || GenUtils.get_account_id(source_id) <> env_dataset_id_append()
 
+    {:ok, table} = get_table(source_id)
+    timepartitioning_type = table.timePartitioning.type
+
     partitioning = %Model.TimePartitioning{
-      type: "DAY",
+      # use the same type as the existing table, as BQ does not allow changing it
+      type: timepartitioning_type,
       expirationMs: table_ttl
     }
 
@@ -147,6 +151,23 @@ defmodule Logflare.Google.BigQuery do
     |> GenUtils.maybe_parse_google_api_result()
   end
 
+  @spec patch_table_clustering(atom, list(String.t()), binary, binary) :: ok_err_tup
+  def patch_table_clustering(source_id, clustering, dataset_id, project_id) do
+    conn = GenUtils.get_conn()
+    table_name = GenUtils.format_table_name(source_id)
+    dataset_id = dataset_id || GenUtils.get_account_id(source_id) <> env_dataset_id_append()
+
+    clustering_model = %Model.Clustering{
+      fields: clustering
+    }
+
+    conn
+    |> Api.Tables.bigquery_tables_patch(project_id, dataset_id, table_name,
+      body: %Model.Table{clustering: clustering_model}
+    )
+    |> GenUtils.maybe_parse_google_api_result()
+  end
+
   @spec patch_table(atom, any, binary, binary) :: {:error, Tesla.Env.t()} | {:ok, Model.Table.t()}
   def patch_table(source_id, schema, dataset_id, project_id) do
     conn = GenUtils.get_conn()

lib/logflare/sources.ex

@@ -150,6 +150,20 @@ defmodule Logflare.Sources do
       )
     end
 
+    if source.bigquery_clustering_fields != updated.bigquery_clustering_fields and
+         not SingleTenant.postgres_backend?() do
+      user = Users.Cache.get(updated.user_id)
+
+      fields = String.split(updated.bigquery_clustering_fields || "", ",") ++ ["timestamp", "id"]
+
+      BigQuery.patch_table_clustering(
+        updated.token,
+        fields,
+        user.bigquery_dataset_id,
+        user.bigquery_project_id
+      )
+    end
+
     {:ok, updated}
   end
 

lib/logflare/sources/source.ex

@@ -29,7 +29,8 @@ defmodule Logflare.Source do
              :custom_event_message_keys,
              :backends,
              :retention_days,
-             :transform_copy_fields
+             :transform_copy_fields,
+             :bigquery_clustering_fields
            ]}
   defp env_dataset_id_append,
     do: Application.get_env(:logflare, Logflare.Google)[:dataset_id_append]
@@ -134,6 +135,7 @@ defmodule Logflare.Source do
     field(:suggested_keys, :string, default: "")
     field(:retention_days, :integer, virtual: true)
     field(:transform_copy_fields, :string)
+    field(:bigquery_clustering_fields, :string)
     # Causes a shitstorm
     # field :bigquery_schema, Ecto.Term
 
@@ -171,6 +173,7 @@ defmodule Logflare.Source do
       :public_token,
       :favorite,
       :bigquery_table_ttl,
+      :bigquery_clustering_fields,
       # users can't update thier API quota currently
       :api_quota,
       :webhook_notification_url,
@@ -202,6 +205,7 @@ defmodule Logflare.Source do
       :public_token,
       :favorite,
       :bigquery_table_ttl,
+      :bigquery_clustering_fields,
       :webhook_notification_url,
       :slack_hook_url,
       :custom_event_message_keys,

lib/logflare_web/templates/source/edit.html.eex

@@ -261,21 +261,35 @@
 
   <%= section_header("Backend TTL") %>
   <%= if @conn.assigns.user.billing_enabled do %>
-  <p>Set how long to keep data in your backend.</p>
+    <p>Set how long to keep data in your backend.</p>
+    <%= form_for @changeset, Routes.source_path(@conn, :update, @source), fn e -> %>
+    <div class="form-group">
+      <%= text_input e, :retention_days, placeholder: "3", class: "form-control form-control-margin" %>
+      <%= error_tag e, :retention_days %>
+      <small class="form-text text-muted">
+        Days to keep data.
+      </small>
+    </div>
+    <%= submit "Update backend settings", class: "btn btn-primary form-button" %>
+    <% end %>
+  <% else %>
+    <p>Add a Google Cloud Platform project ID to get started!</p>
+    <%= link "Setup custom backend", to: Routes.user_path(@conn, :edit) <> "#bigquery-backend", class: "btn btn-primary form-button", role: "button" %>
+  <% end %>
+
+  <%= section_header("Backend Clustering") %>
+  <p>Set additional clustering columns to improve query performance. <code>timestamp</code> and <code>id</code> are always clustered.</p>
   <%= form_for @changeset, Routes.source_path(@conn, :update, @source), fn e -> %>
   <div class="form-group">
-    <%= text_input e, :retention_days, placeholder: "3", class: "form-control form-control-margin" %>
-    <%= error_tag e, :retention_days %>
+    <%= text_input e, :bigquery_clustering_fields, placeholder: "my_field,my_other_field", class: "form-control form-control-margin" %>
+    <%= error_tag e, :bigquery_clustering_fields %>
     <small class="form-text text-muted">
-      Days to keep data.
+      Comma-separated. For example: <code>my_field,my_other_field</code>.
     </small>
   </div>
-  <%= submit "Update backend settings", class: "btn btn-primary form-button" %>
-  <% end %>
-  <% else %>
-  <p>Add a Google Cloud Platform project ID to get started!</p>
-  <%= link "Setup custom backend", to: Routes.user_path(@conn, :edit) <> "#bigquery-backend", class: "btn btn-primary form-button", role: "button" %>
+  <%= submit "Update clustering", class: "btn btn-primary form-button" %>
   <% end %>
+
   <%= section_header("Suggested Search Keys") %>
   <p>Set suggested search keys for this source, to encourage usage of certain filtering keys when performing searches across large sources. If a declared key is missing, users will be prompted to include the key when a search is made.</p>
   <%= form_for @changeset, Routes.source_path(@conn, :update, @source),fn f -> %>

priv/repo/migrations/20250702081228_add_bigquery_clustering_fieldsto_sources_table.exs

@@ -0,0 +1,9 @@
+defmodule Logflare.Repo.Migrations.AddBigqueryClusteringFieldstoSourcesTable do
+  use Ecto.Migration
+
+  def change do
+    alter table(:sources) do
+      add :bigquery_clustering_fields, :string
+    end
+  end
+end