chore: release v3.6.0

* take CLICKHOUSE_DB environment variable into account

Fixes: #4888
Signed-off-by: Aleksandar Pesic <peske.nis@gmail.com>

* chore: cleanup env usage

* chore: extend env.prod.example

---------

Signed-off-by: Aleksandar Pesic <peske.nis@gmail.com>
Co-authored-by: Aleksandar Pesic <peske.nis@gmail.com>

.env.prod.example

@@ -245,6 +245,7 @@ OTEL_SERVICE_NAME="langfuse"
 # CLICKHOUSE_URL=
 # CLICKHOUSE_USER=
 # CLICKHOUSE_PASSWORD=
+# CLICKHOUSE_DB=
 
 # Ingestion
 # LANGFUSE_INGESTION_QUEUE_DELAY_MS=

ee/package.json

@@ -47,7 +47,8 @@
   },
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "langfuse",
-  "version": "3.5.2",
+  "version": "3.6.0",
   "author": "engineering@langfuse.com",
   "license": "MIT",
   "private": true,
@@ -84,7 +84,8 @@
   "packageManager": "pnpm@9.5.0",
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

packages/shared/clickhouse/scripts/down.sh

@@ -18,21 +18,31 @@ then
     exit 1
 fi
 
+# Ensure CLICKHOUSE_DB is set
+if [ -z "${CLICKHOUSE_DB}" ]; then
+    export CLICKHOUSE_DB="default"
+fi
+
+# Ensure CLICKHOUSE_CLUSTER_NAME is set
+if [ -z "${CLICKHOUSE_CLUSTER_NAME}" ]; then
+    export CLICKHOUSE_CLUSTER_NAME="default"
+fi
+
 # Construct the database URL
 if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
   fi
 
   # Execute the up command
   migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" down
 else
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   fi
 
   # Execute the up command

packages/shared/clickhouse/scripts/drop.sh

@@ -12,11 +12,16 @@ then
     exit 1
 fi
 
+# Ensure CLICKHOUSE_DB is set
+if [ -z "${CLICKHOUSE_DB}" ]; then
+    export CLICKHOUSE_DB="default"
+fi
+
 # Construct the database URL
 if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
+    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
 else
-    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
+    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
 fi
 # Execute the drop command
 migrate -source file://clickhouse/migrations -database "$DATABASE_URL" drop

packages/shared/clickhouse/scripts/up.sh

@@ -18,21 +18,31 @@ then
     exit 1
 fi
 
+# Ensure CLICKHOUSE_DB is set
+if [ -z "${CLICKHOUSE_DB}" ]; then
+    export CLICKHOUSE_DB="default"
+fi
+
+# Ensure CLICKHOUSE_CLUSTER_NAME is set
+if [ -z "${CLICKHOUSE_CLUSTER_NAME}" ]; then
+    export CLICKHOUSE_CLUSTER_NAME="default"
+fi
+
 # Construct the database URL
 if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
   fi
 
   # Execute the up command
   migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" up
 else
 if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   fi
 
   # Execute the up command

packages/shared/package.json

@@ -121,7 +121,8 @@
   },
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

packages/shared/src/env.ts

@@ -29,6 +29,8 @@ const EnvSchema = z.object({
   LANGFUSE_CACHE_PROMPT_ENABLED: z.enum(["true", "false"]).default("false"),
   LANGFUSE_CACHE_PROMPT_TTL_SECONDS: z.coerce.number().default(60 * 60),
   CLICKHOUSE_URL: z.string().url(),
+  CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
+  CLICKHOUSE_DB: z.string().default("default"),
   CLICKHOUSE_USER: z.string(),
   CLICKHOUSE_PASSWORD: z.string(),
   LANGFUSE_SDK_CI_SYNC_PROCESSING_ENABLED: z

packages/shared/src/server/clickhouse/client.ts

@@ -10,7 +10,7 @@ export const clickhouseClient = (opts?: NodeClickHouseClientConfigOptions) =>
     url: env.CLICKHOUSE_URL,
     username: env.CLICKHOUSE_USER,
     password: env.CLICKHOUSE_PASSWORD,
-    database: "default",
+    database: env.CLICKHOUSE_DB,
     clickhouse_settings: {
       async_insert: 1,
       wait_for_async_insert: 1, // if disabled, we won't get errors from clickhouse

packages/shared/src/server/repositories/dashboards.ts

@@ -418,7 +418,7 @@ export const getObservationLatencies = async (
   // Skipping FINAL here, as the quantiles are approximate to begin with.
   const query = `
     SELECT
-      quantiles(0.5, 0.9, 0.95, 0.99)(date_diff('milliseconds', o.start_time, o.end_time)) as quantiles,
+      quantiles(0.5, 0.9, 0.95, 0.99)(date_diff('millisecond', o.start_time, o.end_time)) as quantiles,
       name
     FROM observations o
     ${chFilter.find((f) => f.clickhouseTable === "traces") ? "LEFT JOIN traces t ON o.trace_id = t.id AND o.project_id = t.project_id" : ""}
@@ -465,7 +465,7 @@ export const getTracesLatencies = async (
       select o.trace_id,
              t.name,
              o.project_id,
-             date_diff('milliseconds', min(o.start_time), coalesce(max(o.end_time), max(o.start_time))) as duration
+             date_diff('millisecond', min(o.start_time), coalesce(max(o.end_time), max(o.start_time))) as duration
       FROM traces t 
       JOIN observations o
       ON o.trace_id = t.id AND o.project_id = t.project_id
@@ -521,7 +521,7 @@ export const getModelLatenciesOverTime = async (
   SELECT 
     ${selectTimeseriesColumn(groupBy, "o.start_time", "start_time_bucket")},
     provided_model_name,
-    quantiles(0.5, 0.75, 0.9, 0.95, 0.99)(date_diff('milliseconds', o.start_time, o.end_time)) as quantiles
+    quantiles(0.5, 0.75, 0.9, 0.95, 0.99)(date_diff('millisecond', o.start_time, o.end_time)) as quantiles
   FROM observations o
   ${traceFilter ? "JOIN traces t ON o.trace_id = t.id AND o.project_id = t.project_id" : ""}
   WHERE project_id = {projectId: String}

packages/shared/src/server/repositories/observations.ts

@@ -540,8 +540,8 @@ const getObservationsTableInternal = async <T>(
         o.prompt_name as "prompt_name",
         o.prompt_version as "prompt_version",
         internal_model_id as "internal_model_id",
-        if(isNull(end_time), NULL, date_diff('milliseconds', start_time, end_time)) as latency,
-        if(isNull(completion_start_time), NULL,  date_diff('milliseconds', start_time, completion_start_time)) as "time_to_first_token"`;
+        if(isNull(end_time), NULL, date_diff('millisecond', start_time, end_time)) as latency,
+        if(isNull(completion_start_time), NULL,  date_diff('millisecond', start_time, completion_start_time)) as "time_to_first_token"`;
 
   const { projectId, filter, selectIOAndMetadata, limit, offset, orderBy } =
     opts;
@@ -1010,7 +1010,7 @@ export const getObservationMetricsForPrompts = async (
                   end_time,
                   usage_details,
                   cost_details,
-                  dateDiff('milliseconds', start_time, end_time) AS latency_ms
+                  dateDiff('millisecond', start_time, end_time) AS latency_ms
               FROM observations
               FINAL
               WHERE (type = 'GENERATION') 
@@ -1073,7 +1073,7 @@ export const getLatencyAndTotalCostForObservations = async (
     SELECT
         id,
         cost_details['total'] AS total_cost,
-        dateDiff('milliseconds', start_time, end_time) AS latency_ms
+        dateDiff('millisecond', start_time, end_time) AS latency_ms
     FROM observations FINAL 
     WHERE project_id = {projectId: String} 
     AND id IN ({observationIds: Array(String)})
@@ -1105,7 +1105,7 @@ export const getLatencyAndTotalCostForObservationsByTraces = async (
     SELECT
         trace_id,
         sumMap(cost_details)['total'] AS total_cost,
-        dateDiff('milliseconds', min(start_time), max(end_time)) AS latency_ms
+        dateDiff('millisecond', min(start_time), max(end_time)) AS latency_ms
     FROM observations FINAL
     WHERE project_id = {projectId: String} 
     AND trace_id IN ({traceIds: Array(String)})
@@ -1225,12 +1225,12 @@ export const getGenerationsForPostHog = async (
       o.start_time as start_time,
       o.id as id,
       o.total_cost as total_cost,
-      if(isNull(completion_start_time), NULL, date_diff('milliseconds', start_time, completion_start_time)) as time_to_first_token,
+      if(isNull(completion_start_time), NULL, date_diff('millisecond', start_time, completion_start_time)) as time_to_first_token,
       o.usage_details['total'] as input_tokens,
       o.usage_details['output'] as output_tokens,
       o.cost_details['total'] as total_tokens,
       o.project_id as project_id,
-      if(isNull(end_time), NULL, date_diff('milliseconds', start_time, end_time) / 1000) as latency,
+      if(isNull(end_time), NULL, date_diff('millisecond', start_time, end_time) / 1000) as latency,
       o.provided_model_name as model,
       o.level as level,
       o.version as version,

packages/shared/src/server/repositories/traces.ts

@@ -526,7 +526,7 @@ const getSessionsTableGeneric = async <T>(props: FetchSessionsTableProps) => {
             groupUniqArrayArray(t.tags) as trace_tags,
             -- Aggregate observations data at session level
             sum(o.obs_count) as total_observations,
-            date_diff('milliseconds', min(min_start_time), max(max_end_time)) as duration,
+            date_diff('millisecond', min(min_start_time), max(max_end_time)) as duration,
             sumMap(o.sum_usage_details) as session_usage_details,
             sumMap(o.sum_cost_details) as session_cost_details,
             arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_cost_details)))) as session_input_cost,
@@ -831,7 +831,7 @@ export const getTracesForPostHog = async (
              o.trace_id,
              sum(total_cost) as total_cost,
              count(*) as observation_count,
-             date_diff('milliseconds', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds
+             date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds
       FROM observations o FINAL
       WHERE o.project_id = {projectId: String}
       AND o.start_time >= {minTimestamp: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}

packages/shared/src/server/services/traces-ui-table-service.ts

@@ -351,7 +351,7 @@ const getTracesTableGeneric = async <T>(props: FetchTracesTableProps) => {
         COUNT(*) AS observation_count,
           sumMap(usage_details) as usage_details,
           SUM(total_cost) AS total_cost,
-          date_diff('milliseconds', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
+          date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
           multiIf(
             arrayExists(x -> x = 'ERROR', groupArray(level)), 'ERROR',
             arrayExists(x -> x = 'WARNING', groupArray(level)), 'WARNING',

packages/shared/src/tableDefinitions/mapObservationsTable.ts

@@ -68,7 +68,7 @@ export const observationsTableUiColumnDefinitions: UiColumnMapping[] = [
     uiTableId: "timeToFirstToken",
     clickhouseTableName: "observations",
     clickhouseSelect:
-      "if(isNull(completion_start_time), NULL,  date_diff('milliseconds', start_time, completion_start_time) / 1000)",
+      "if(isNull(completion_start_time), NULL,  date_diff('millisecond', start_time, completion_start_time) / 1000)",
     // If we use the default of Decimal64(12), we cannot filter for more than ~40min due to an overflow
     clickhouseTypeOverwrite: "Decimal64(3)",
   },
@@ -77,7 +77,7 @@ export const observationsTableUiColumnDefinitions: UiColumnMapping[] = [
     uiTableId: "latency",
     clickhouseTableName: "observations",
     clickhouseSelect:
-      "if(isNull(end_time), NULL, date_diff('milliseconds', start_time, end_time) / 1000)",
+      "if(isNull(end_time), NULL, date_diff('millisecond', start_time, end_time) / 1000)",
     // If we use the default of Decimal64(12), we cannot filter for more than ~40min due to an overflow
     clickhouseTypeOverwrite: "Decimal64(3)",
   },
@@ -86,7 +86,7 @@ export const observationsTableUiColumnDefinitions: UiColumnMapping[] = [
     uiTableId: "tokensPerSecond",
     clickhouseTableName: "observations",
     clickhouseSelect:
-      "(arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, usage_details))) / (date_diff('milliseconds', start_time, end_time) / 1000))",
+      "(arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, usage_details))) / (date_diff('millisecond', start_time, end_time) / 1000))",
   },
   {
     uiTableName: "Input Cost ($)",

pnpm-lock.yaml

@@ -6,6 +6,7 @@ settings:
 
 overrides:
   jsonpath-plus: 10.0.7
+  nanoid: ^3.3.8
 
 importers:
 
@@ -742,7 +743,7 @@ importers:
         version: 7.12.0(eslint@8.57.0)(typescript@5.4.5)
       autoprefixer:
         specifier: ^10.4.19
-        version: 10.4.19(postcss@8.4.47)
+        version: 10.4.19(postcss@8.4.49)
       dotenv-cli:
         specifier: ^7.4.2
         version: 7.4.2
@@ -761,9 +762,6 @@ importers:
       node-mocks-http:
         specifier: ^1.14.1
         version: 1.14.1
-      postcss:
-        specifier: ^8.4.47
-        version: 8.4.47
       prettier:
         specifier: ^3.3.3
         version: 3.3.3
@@ -8976,13 +8974,8 @@ packages:
   mz@2.7.0:
     resolution: {integrity: sha512-z81GNO7nnYMEhrGh9LeymoE4+Yr0Wn5McHIZMK5cfQCl+NDX08sCZgUc9/6MHni9IWuFLm1Z3HTCXu2z9fN62Q==}
 
-  nanoid@3.3.6:
-    resolution: {integrity: sha512-BGcqMMJuToF7i1rt+2PWSNVnWIkGCU78jBG3RxO/bZlnZPK2Cmi2QaffxGO/2RvWi9sL+FAiRiXMgsyxQ1DIDA==}
-    engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1}
-    hasBin: true
-
-  nanoid@3.3.7:
-    resolution: {integrity: sha512-eSRppjcPIatRIMC1U6UngP8XFcz8MQWGQdt1MTBQ7NaAmvXDfvNxbvWV3x2y6CdEUciCSsDHDQZbhYaB8QEo2g==}
+  nanoid@3.3.8:
+    resolution: {integrity: sha512-WNLf5Sd8oZxOm+TzppcYk8gVOgP+l58xNy58D0nbUnOxOWRWvlcCV4kUF7ltmI6PsrLl/BgKEyS4mqsGChFN0w==}
     engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1}
     hasBin: true
 
@@ -9518,9 +9511,6 @@ packages:
   picocolors@1.0.0:
     resolution: {integrity: sha512-1fygroTLlHu66zi26VoTDv8yRgm0Fccecssto+MhsZ0D/DGW2sm8E8AjW7NU5VVTRt5GxbeZ5qBuJr+HyLYkjQ==}
 
-  picocolors@1.1.0:
-    resolution: {integrity: sha512-TQ92mBOW0l3LeMeyLV6mzy/kWr8lkd/hp3mTg7wYK7zJhuBStmGMBG0BdeDZS/dZx1IukaX6Bk11zcln25o1Aw==}
-
   picocolors@1.1.1:
     resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==}
 
@@ -9612,10 +9602,6 @@ packages:
     resolution: {integrity: sha512-PS08Iboia9mts/2ygV3eLpY5ghnUcfLV/EXTOW1E2qYxJKGGBUtNjN76FYHnMs36RmARn41bC0AZmn+rR0OVpQ==}
     engines: {node: ^10 || ^12 || >=14}
 
-  postcss@8.4.47:
-    resolution: {integrity: sha512-56rxCq7G/XfB4EkXq9Egn5GCqugWvDFjafDOThIdMBsI15iqPqR5r15TfSr1YPYeEI19YeaXMCbY6u88Y76GLQ==}
-    engines: {node: ^10 || ^12 || >=14}
-
   postcss@8.4.49:
     resolution: {integrity: sha512-OCVPnIObs4N29kxTjzLfUryOkvZEq+pf8jTF0lg8E7uETuWHA+v7j3c/xJmiqpX450191LlmZfUKkXxkTry7nA==}
     engines: {node: ^10 || ^12 || >=14}
@@ -11575,7 +11561,7 @@ snapshots:
     dependencies:
       '@ai-sdk/provider': 0.0.24
       eventsource-parser: 1.1.2
-      nanoid: 3.3.6
+      nanoid: 3.3.8
       secure-json-parse: 2.7.0
     optionalDependencies:
       zod: 3.23.8
@@ -17720,7 +17706,7 @@ snapshots:
       eventsource-parser: 1.1.2
       json-schema: 0.4.0
       jsondiffpatch: 0.6.0
-      nanoid: 3.3.6
+      nanoid: 3.3.8
       secure-json-parse: 2.7.0
       zod-to-json-schema: 3.23.2(zod@3.23.8)
     optionalDependencies:
@@ -17933,14 +17919,14 @@ snapshots:
 
   asynckit@0.4.0: {}
 
-  autoprefixer@10.4.19(postcss@8.4.47):
+  autoprefixer@10.4.19(postcss@8.4.49):
     dependencies:
       browserslist: 4.23.0
       caniuse-lite: 1.0.30001599
       fraction.js: 4.3.7
       normalize-range: 0.1.2
       picocolors: 1.0.0
-      postcss: 8.4.47
+      postcss: 8.4.49
       postcss-value-parser: 4.2.0
 
   available-typed-arrays@1.0.7:
@@ -22282,9 +22268,7 @@ snapshots:
       object-assign: 4.1.1
       thenify-all: 1.6.0
 
-  nanoid@3.3.6: {}
-
-  nanoid@3.3.7: {}
+  nanoid@3.3.8: {}
 
   natural-compare@1.4.0: {}
 
@@ -22851,8 +22835,6 @@ snapshots:
 
   picocolors@1.0.0: {}
 
-  picocolors@1.1.0: {}
-
   picocolors@1.1.1: {}
 
   picomatch@2.3.1: {}
@@ -22926,19 +22908,13 @@ snapshots:
 
   postcss@8.4.31:
     dependencies:
-      nanoid: 3.3.7
+      nanoid: 3.3.8
       picocolors: 1.1.1
       source-map-js: 1.2.1
 
-  postcss@8.4.47:
-    dependencies:
-      nanoid: 3.3.7
-      picocolors: 1.1.0
-      source-map-js: 1.2.1
-
   postcss@8.4.49:
     dependencies:
-      nanoid: 3.3.7
+      nanoid: 3.3.8
       picocolors: 1.1.1
       source-map-js: 1.2.1
 

web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "web",
-  "version": "3.5.2",
+  "version": "3.6.0",
   "private": true,
   "license": "MIT",
   "engines": {
@@ -175,7 +175,6 @@
     "jest": "^29.7.0",
     "jest-environment-jsdom": "^29.7.0",
     "node-mocks-http": "^1.14.1",
-    "postcss": "^8.4.47",
     "prettier": "^3.3.3",
     "prettier-plugin-tailwindcss": "^0.6.6",
     "tailwindcss": "^3.4.17",
@@ -193,7 +192,8 @@
   },
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

web/src/components/ModelParameters/index.tsx

@@ -34,15 +34,12 @@ export type ModelParamsContext = {
   modelParamsDescription?: string;
 };
 
-export const ModelParameters: React.FC<
-  ModelParamsContext & { evalModelsOnly: boolean }
-> = ({
+export const ModelParameters: React.FC<ModelParamsContext> = ({
   modelParams,
   availableProviders,
   availableModels,
   updateModelParamValue,
   setModelParamEnabled,
-  evalModelsOnly,
   formDisabled = false,
   modelParamsDescription,
 }) => {
@@ -55,11 +52,8 @@ export const ModelParameters: React.FC<
       <p className="font-semibold">Model</p>
       {availableProviders.length === 0 ? (
         <>
-          <p className="text-xs">
-            No LLM API key set in project.{" "}
-            {evalModelsOnly && "For evals, only OpenAI models are supported."}
-          </p>
-          <CreateLLMApiKeyDialog evalModelsOnly={evalModelsOnly} />
+          <p className="text-xs">No LLM API key set in project. </p>
+          <CreateLLMApiKeyDialog />
         </>
       ) : (
         <div className="space-y-4">
@@ -70,7 +64,6 @@ export const ModelParameters: React.FC<
             value={modelParams.provider.value}
             options={availableProviders}
             updateModelParam={updateModelParamValue}
-            evalModelsOnly={evalModelsOnly}
           />
           <ModelParamsSelect
             title="Model name"
@@ -149,7 +142,6 @@ type ModelParamsSelectProps = {
   updateModelParam: ModelParamsContext["updateModelParamValue"];
   disabled?: boolean;
   modelParamsDescription?: string;
-  evalModelsOnly?: boolean;
 };
 const ModelParamsSelect = ({
   title,
@@ -159,7 +151,6 @@ const ModelParamsSelect = ({
   updateModelParam,
   disabled,
   modelParamsDescription,
-  evalModelsOnly,
 }: ModelParamsSelectProps) => {
   return (
     <div className="space-y-2">
@@ -191,7 +182,7 @@ const ModelParamsSelect = ({
             </SelectItem>
           ))}
           <SelectSeparator />
-          <CreateLLMApiKeyDialog evalModelsOnly={evalModelsOnly} />
+          <CreateLLMApiKeyDialog />
         </SelectContent>
       </Select>
       {modelParamsDescription ? (

web/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v3.5.2";
+export const VERSION = "v3.6.0";

web/src/ee/features/evals/components/template-form.tsx

@@ -545,7 +545,6 @@ export const InnerEvalTemplateForm = (props: {
                 modelParamsDescription:
                   "Select a model which supports function calling.",
               }}
-              evalModelsOnly
               formDisabled={!props.isEditing}
             />
           </div>

web/src/ee/features/experiments/components/CreateExperimentsForm.tsx

@@ -683,10 +683,7 @@ export const CreateExperimentsForm = ({
                       availableProviders,
                       updateModelParamValue: updateModelParamValue,
                       setModelParamEnabled,
-                      modelParamsDescription:
-                        "Select a model which supports function calling.",
                     }}
-                    evalModelsOnly
                   />
                 </Card>
                 {form.formState.errors.modelConfig && (

web/src/ee/features/playground/page/playground.tsx

@@ -14,7 +14,7 @@ export default function Playground() {
       <div className="max-h-full min-h-0 basis-1/4 pr-2">
         <div className="grid h-full grid-rows-[minmax(20dvh,max-content),minmax(20dvh,auto)] overflow-auto">
           <div className="mb-4 max-h-[80dvh] min-h-[20dvh] overflow-y-auto">
-            <ModelParameters {...playgroundContext} evalModelsOnly={false} />
+            <ModelParameters {...playgroundContext} />
           </div>
           <div className="min-h-[20dvh]">
             <Variables />

web/src/env.mjs

@@ -150,6 +150,8 @@ export const env = createEnv({
 
     // clickhouse
     CLICKHOUSE_URL: z.string().url(),
+    CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
+    CLICKHOUSE_DB: z.string().default("default"),
     CLICKHOUSE_USER: z.string(),
     CLICKHOUSE_PASSWORD: z.string(),
     CLICKHOUSE_CLUSTER_ENABLED: z.enum(["true", "false"]).default("false"),
@@ -378,12 +380,10 @@ export const env = createEnv({
     AUTH_CUSTOM_ISSUER: process.env.AUTH_CUSTOM_ISSUER,
     AUTH_CUSTOM_NAME: process.env.AUTH_CUSTOM_NAME,
     AUTH_CUSTOM_SCOPE: process.env.AUTH_CUSTOM_SCOPE,
-    AUTH_CUSTOM_CLIENT_AUTH_METHOD:
-      process.env.AUTH_CUSTOM_CLIENT_AUTH_METHOD,
+    AUTH_CUSTOM_CLIENT_AUTH_METHOD: process.env.AUTH_CUSTOM_CLIENT_AUTH_METHOD,
     AUTH_CUSTOM_ALLOW_ACCOUNT_LINKING:
       process.env.AUTH_CUSTOM_ALLOW_ACCOUNT_LINKING,
-    AUTH_IGNORE_ACCOUNT_FIELDS:
-      process.env.AUTH_IGNORE_ACCOUNT_FIELDS,
+    AUTH_IGNORE_ACCOUNT_FIELDS: process.env.AUTH_IGNORE_ACCOUNT_FIELDS,
     AUTH_DOMAINS_WITH_SSO_ENFORCEMENT:
       process.env.AUTH_DOMAINS_WITH_SSO_ENFORCEMENT,
     AUTH_DISABLE_USERNAME_PASSWORD: process.env.AUTH_DISABLE_USERNAME_PASSWORD,
@@ -445,6 +445,8 @@ export const env = createEnv({
     NEXT_PUBLIC_CRISP_WEBSITE_ID: process.env.NEXT_PUBLIC_CRISP_WEBSITE_ID,
     // clickhouse
     CLICKHOUSE_URL: process.env.CLICKHOUSE_URL,
+    CLICKHOUSE_CLUSTER_NAME: process.env.CLICKHOUSE_CLUSTER_NAME,
+    CLICKHOUSE_DB: process.env.CLICKHOUSE_DB,
     CLICKHOUSE_USER: process.env.CLICKHOUSE_USER,
     CLICKHOUSE_PASSWORD: process.env.CLICKHOUSE_PASSWORD,
     CLICKHOUSE_CLUSTER_ENABLED: process.env.CLICKHOUSE_CLUSTER_ENABLED,

web/src/features/datasets/components/DatasetCompareRunsTable.tsx

@@ -26,7 +26,7 @@ import { DatasetCompareRunPeekView } from "@/src/features/datasets/components/Da
 import { useClickhouse } from "@/src/components/layouts/ClickhouseAdminToggle";
 import { getQueryKey } from "@trpc/react-query";
 import { useQueryClient } from "@tanstack/react-query";
-
+import _ from "lodash";
 export type RunMetrics = {
   id: string;
   scores: ScoreAggregate;
@@ -182,7 +182,7 @@ export function DatasetCompareRunsTable(props: {
           ({ datasetItemId, trace, observation, scores }) => {
             if (!itemsAcc[datasetItemId]) itemsAcc[datasetItemId] = {};
 
-            itemsAcc[datasetItemId][runId] = {
+            _.set(itemsAcc[datasetItemId], runId, {
               id: runId,
               traceId: trace?.id ?? "",
               observationId: observation?.id ?? undefined,
@@ -196,7 +196,7 @@ export function DatasetCompareRunsTable(props: {
                     : usdFormatter(trace?.totalCost)) ?? undefined,
               },
               scores,
-            };
+            });
           },
         );
 

web/src/features/datasets/server/service.ts

@@ -176,7 +176,7 @@ export const createTempTableInClickhouse = async (
   clickhouseSession: string,
 ) => {
   const query = `
-      CREATE TABLE IF NOT EXISTS ${tableName} ${env.CLICKHOUSE_CLUSTER_ENABLED === "true" ? "ON CLUSTER default" : ""}
+      CREATE TABLE IF NOT EXISTS ${tableName} ${env.CLICKHOUSE_CLUSTER_ENABLED === "true" ? "ON CLUSTER " + env.CLICKHOUSE_CLUSTER_NAME : ""}
       (
           project_id String,    
           run_id String,  
@@ -289,7 +289,7 @@ const getObservationLatencyAndCostForDataset = async (
   const query = `
     WITH agg AS (
       SELECT
-          dateDiff('milliseconds', start_time, end_time) AS latency_ms,
+          dateDiff('millisecond', start_time, end_time) AS latency_ms,
           total_cost AS cost,
           run_id
       FROM observations AS o
@@ -342,7 +342,7 @@ const getTraceLatencyAndCostForDataset = async (
       SELECT
         o.trace_id,
         run_id,
-        dateDiff('milliseconds', min(start_time), max(end_time)) AS latency_ms,
+        dateDiff('millisecond', min(start_time), max(end_time)) AS latency_ms,
         sum(total_cost) AS cost
       FROM observations o JOIN ${tableName} tmp
         ON tmp.project_id = o.project_id 

web/src/features/models/validation.ts

@@ -17,9 +17,10 @@ export const GetModelResultSchema = z.object({
   projectId: z.string().nullable(),
   modelName: z.string(),
   matchPattern: z.string(),
-  tokenizerConfig: z
-    .record(z.union([z.string(), z.coerce.number()]))
-    .nullable(),
+  tokenizerConfig: z.union([
+    z.record(z.union([z.string(), z.coerce.number()])).nullable(),
+    z.string(),
+  ]),
   tokenizerId: TokenizerSchema,
   prices: PriceMapSchema,
 });

web/src/features/public-api/components/CreateLLMApiKeyDialog.tsx

@@ -14,11 +14,7 @@ import useProjectIdFromURL from "@/src/hooks/useProjectIdFromURL";
 import { useUiCustomization } from "@/src/ee/features/ui-customization/useUiCustomization";
 import { CreateLLMApiKeyForm } from "@/src/features/public-api/components/CreateLLMApiKeyForm";
 
-export function CreateLLMApiKeyDialog({
-  evalModelsOnly,
-}: {
-  evalModelsOnly?: boolean;
-}) {
+export function CreateLLMApiKeyDialog() {
   const projectId = useProjectIdFromURL();
   const [open, setOpen] = useState(false);
   const hasAccess = useHasProjectAccess({
@@ -49,7 +45,6 @@ export function CreateLLMApiKeyDialog({
         {open && (
           <CreateLLMApiKeyForm
             projectId={projectId}
-            evalModelsOnly={evalModelsOnly}
             onSuccess={() => setOpen(false)}
             customization={uiCustomization}
           />

web/src/features/public-api/components/CreateLLMApiKeyForm.tsx

@@ -66,12 +66,10 @@ const formSchema = z
 
 export function CreateLLMApiKeyForm({
   projectId,
-  evalModelsOnly,
   onSuccess,
   customization,
 }: {
   projectId?: string;
-  evalModelsOnly?: boolean;
   onSuccess: () => void;
   customization: ReturnType<typeof useUiCustomization>;
 }) {
@@ -248,18 +246,11 @@ export function CreateLLMApiKeyForm({
                   </SelectTrigger>
                 </FormControl>
                 <SelectContent>
-                  {Object.values(LLMAdapter)
-                    .filter(
-                      (provider) =>
-                        !evalModelsOnly ||
-                        provider === LLMAdapter.OpenAI ||
-                        provider === LLMAdapter.Azure,
-                    )
-                    .map((provider) => (
-                      <SelectItem value={provider} key={provider}>
-                        {provider}
-                      </SelectItem>
-                    ))}
+                  {Object.values(LLMAdapter).map((provider) => (
+                    <SelectItem value={provider} key={provider}>
+                      {provider}
+                    </SelectItem>
+                  ))}
                 </SelectContent>
               </Select>
               <FormMessage />

web/src/features/public-api/server/traces.ts

@@ -58,7 +58,7 @@ export const generateTracesForPublicApi = async (
         trace_id,
         project_id,
         sum(total_cost) as total_cost,
-        date_diff('milliseconds', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
+        date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
         groupArray(id) as observation_ids
       FROM observations FINAL
       WHERE project_id = {projectId: String}

web/src/pages/api/public/media/[mediaId].ts

@@ -73,6 +73,7 @@ export default withMiddlewares({
     }),
     bodySchema: PatchMediaBodySchema,
     responseSchema: z.void(),
+    rateLimitResource: "ingestion",
     fn: async ({ query, body, auth }) => {
       if (auth.scope.accessLevel !== "all") throw new ForbiddenError();
 

web/src/pages/api/public/media/index.ts

@@ -26,6 +26,7 @@ export default withMiddlewares({
     bodySchema: GetMediaUploadUrlQuerySchema,
     responseSchema: GetMediaUploadUrlResponseSchema,
     successStatusCode: 201,
+    rateLimitResource: "ingestion",
     fn: async ({ body, auth }) => {
       if (auth.scope.accessLevel !== "all") throw new ForbiddenError();
 

web/src/pages/api/public/scores/[scoreId].ts

@@ -1,4 +1,3 @@
-import { env } from "@/src/env.mjs";
 import { createAuthedAPIRoute } from "@/src/features/public-api/server/createAuthedAPIRoute";
 import { withMiddlewares } from "@/src/features/public-api/server/withMiddlewares";
 import { measureAndReturnApi } from "@/src/server/utils/checkClickhouseAccess";
@@ -65,32 +64,15 @@ export default withMiddlewares({
     fn: async ({ query, auth }) => {
       const { scoreId } = query;
 
-      const score = await prisma.score.findUnique({
-        select: {
-          id: true,
-        },
-        where: {
-          id: scoreId,
-          projectId: auth.scope.projectId,
-        },
-      });
-
-      if (!score) {
-        throw new LangfuseNotFoundError(
-          "Score not found within authorized project",
-        );
-      }
-
-      if (env.CLICKHOUSE_URL) {
-        await deleteScore(auth.scope.projectId, scoreId);
-      }
-
-      await prisma.score.delete({
-        where: {
-          id: scoreId,
-          projectId: auth.scope.projectId,
-        },
-      });
+      await Promise.all([
+        prisma.score.deleteMany({
+          where: {
+            id: scoreId,
+            projectId: auth.scope.projectId,
+          },
+        }),
+        deleteScore(auth.scope.projectId, scoreId),
+      ]);
 
       return { message: "Score deleted successfully" };
     },

web/src/server/api/routers/comments.ts

@@ -11,7 +11,10 @@ import { auditLog } from "@/src/features/audit-logs/auditLog";
 import { TRPCError } from "@trpc/server";
 import { validateCommentReferenceObject } from "@/src/features/comments/validateCommentReferenceObject";
 import { measureAndReturnApi } from "@/src/server/utils/checkClickhouseAccess";
-import { getTracesIdentifierForSession } from "@langfuse/shared/src/server";
+import {
+  getTracesIdentifierForSession,
+  logger,
+} from "@langfuse/shared/src/server";
 
 export const commentsRouter = createTRPCRouter({
   create: protectedProjectProcedure
@@ -56,7 +59,7 @@ export const commentsRouter = createTRPCRouter({
 
         return comment;
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.create", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -115,7 +118,7 @@ export const commentsRouter = createTRPCRouter({
           before: comment,
         });
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.delete", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -172,7 +175,7 @@ export const commentsRouter = createTRPCRouter({
 
         return comments;
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.getByObjectId", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -207,7 +210,7 @@ export const commentsRouter = createTRPCRouter({
         });
         return new Map([[input.objectId, commentCount]]);
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.getCountByObjectId", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -251,7 +254,7 @@ export const commentsRouter = createTRPCRouter({
           ]),
         );
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.getCountByObjectType", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -339,8 +342,11 @@ export const commentsRouter = createTRPCRouter({
             );
           },
         });
-      } catch (e) {
-        console.error(e);
+      } catch (error) {
+        logger.error(
+          "Failed to call comments.getTraceCommentCountBySessionId",
+          error,
+        );
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "Unable to get trace comment counts by session id",

web/src/server/api/routers/sessions.ts

@@ -32,6 +32,7 @@ import {
   getCostForTraces,
   getTracesGroupedByUsers,
   getPublicSessionsFilter,
+  logger,
 } from "@langfuse/shared/src/server";
 
 const SessionFilterOptions = z.object({
@@ -129,7 +130,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to call sessions.all", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get sessions",
@@ -196,7 +197,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Error in sessions.countAll", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session count",
@@ -277,7 +278,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Error in sessions.metrics", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session metrics",
@@ -391,7 +392,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to get sessions.filterOptions", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session filter options",
@@ -549,7 +550,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to get sessions.byId", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session",
@@ -593,7 +594,7 @@ export const sessionRouter = createTRPCRouter({
         });
         return session;
       } catch (error) {
-        console.error(error);
+        logger.error("Unable to call sessions.bookmark", error);
         if (
           error instanceof Prisma.PrismaClientKnownRequestError &&
           error.code === "P2025" // Record to update not found
@@ -643,7 +644,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to call sessions.publish", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to publish session",

web/src/server/api/routers/traces.ts

@@ -769,7 +769,7 @@ export const traceRouter = createTRPCRouter({
 
         return trace;
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call traces.bookmark", error);
         if (
           error instanceof Prisma.PrismaClientKnownRequestError &&
           error.code === "P2025" // Record to update not found
@@ -839,7 +839,7 @@ export const traceRouter = createTRPCRouter({
           return clickhouseTrace;
         }
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call traces.publish", error);
         if (
           error instanceof Prisma.PrismaClientKnownRequestError &&
           error.code === "P2025" // Record to update not found
@@ -907,7 +907,7 @@ export const traceRouter = createTRPCRouter({
           await upsertTrace(convertTraceDomainToClickhouse(clickhouseTrace));
         }
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call traces.updateTags", error);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
         });

worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "worker",
-  "version": "3.5.2",
+  "version": "3.6.0",
   "description": "",
   "license": "MIT",
   "private": true,

worker/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v3.5.2";
+export const VERSION = "v3.6.0";

worker/src/env.ts

@@ -82,6 +82,8 @@ const EnvSchema = z.object({
 
   CLICKHOUSE_URL: z.string().url(),
   CLICKHOUSE_USER: z.string(),
+  CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
+  CLICKHOUSE_DB: z.string().default("default"),
   CLICKHOUSE_PASSWORD: z.string(),
 
   LANGFUSE_LEGACY_INGESTION_WORKER_CONCURRENCY: z.coerce

worker/src/queues/evalQueue.ts

@@ -1,5 +1,5 @@
 import { Job } from "bullmq";
-import { BaseError } from "@langfuse/shared";
+import { ApiError, BaseError } from "@langfuse/shared";
 import { kyselyPrisma } from "@langfuse/shared/src/db";
 import { sql } from "kysely";
 import {
@@ -64,22 +64,21 @@ export const evalJobExecutorQueueProcessor = async (
 
     // do not log expected errors (api failures + missing api keys not provided by the user)
     if (
-      !(e instanceof BaseError && e.message.includes("API key for provider")) &&
-      !(
-        e instanceof BaseError &&
+      (e instanceof BaseError && e.message.includes("API key for provider")) || // api key not provided
+      (e instanceof ApiError && e.httpCode >= 400 && e.httpCode < 500) || // do not error and retry on 4xx errors. They are visible to the user in the UI but do not alert us.
+      (e instanceof BaseError &&
         e.message.includes(
           "Please ensure the mapped data exists and consider extending the job delay.",
-        )
-      )
+        )) // Trace not found.
     ) {
-      traceException(e);
-      logger.error(
-        `Failed Evaluation_Execution job for id ${job.data.payload.jobExecutionId}`,
-        e,
-      );
-      throw e;
+      return;
     }
 
-    return;
+    traceException(e);
+    logger.error(
+      `Failed Evaluation_Execution job for id ${job.data.payload.jobExecutionId}`,
+      e,
+    );
+    throw e;
   }
 };

worker/src/services/IngestionService/index.ts

@@ -9,9 +9,6 @@ import {
   convertObservationReadToInsert,
   convertScoreReadToInsert,
   convertTraceReadToInsert,
-  convertPostgresObservationToInsert,
-  convertPostgresScoreToInsert,
-  convertPostgresTraceToInsert,
   eventTypes,
   findModel,
   IngestionEventType,
@@ -135,59 +132,47 @@ export class IngestionService {
       minTimestamp === Infinity
         ? undefined
         : convertDateToClickhouseDateTime(new Date(minTimestamp));
-    const [postgresScoreRecord, clickhouseScoreRecord, scoreRecords] =
-      await Promise.all([
-        this.getPostgresRecord({
-          projectId,
-          entityId,
-          table: TableName.Scores,
-        }),
-        this.getClickhouseRecord({
-          projectId,
-          entityId,
-          table: TableName.Scores,
-          additionalFilters: {
-            whereCondition: timestamp
-              ? " AND timestamp >= {timestamp: DateTime64(3)} "
-              : "",
-            params: { timestamp },
-          },
-        }),
-        Promise.all(
-          timeSortedEvents.map(async (scoreEvent) => {
-            const validatedScore = await validateAndInflateScore({
-              body: scoreEvent.body,
-              scoreId: entityId,
-              projectId,
-            });
+    const [clickhouseScoreRecord, scoreRecords] = await Promise.all([
+      this.getClickhouseRecord({
+        projectId,
+        entityId,
+        table: TableName.Scores,
+        additionalFilters: {
+          whereCondition: timestamp
+            ? " AND timestamp >= {timestamp: DateTime64(3)} "
+            : "",
+          params: { timestamp },
+        },
+      }),
+      Promise.all(
+        timeSortedEvents.map(async (scoreEvent) => {
+          const validatedScore = await validateAndInflateScore({
+            body: scoreEvent.body,
+            scoreId: entityId,
+            projectId,
+          });
 
-            return {
-              id: entityId,
-              project_id: projectId,
-              timestamp: this.getMillisecondTimestamp(scoreEvent.timestamp),
-              name: validatedScore.name,
-              value: validatedScore.value,
-              source: validatedScore.source,
-              trace_id: validatedScore.traceId,
-              data_type: validatedScore.dataType,
-              observation_id: validatedScore.observationId,
-              comment: validatedScore.comment,
-              string_value: validatedScore.stringValue,
-              created_at: Date.now(),
-              updated_at: Date.now(),
-              event_ts: new Date(scoreEvent.timestamp).getTime(),
-              is_deleted: 0,
-            };
-          }),
-        ),
-      ]);
+          return {
+            id: entityId,
+            project_id: projectId,
+            timestamp: this.getMillisecondTimestamp(scoreEvent.timestamp),
+            name: validatedScore.name,
+            value: validatedScore.value,
+            source: validatedScore.source,
+            trace_id: validatedScore.traceId,
+            data_type: validatedScore.dataType,
+            observation_id: validatedScore.observationId,
+            comment: validatedScore.comment,
+            string_value: validatedScore.stringValue,
+            created_at: Date.now(),
+            updated_at: Date.now(),
+            event_ts: new Date(scoreEvent.timestamp).getTime(),
+            is_deleted: 0,
+          };
+        }),
+      ),
+    ]);
 
-    if (postgresScoreRecord) {
-      recordIncrement("langfuse.ingestion.lookup.hit", 1, {
-        store: "postgres",
-        object: "score",
-      });
-    }
     if (clickhouseScoreRecord) {
       recordIncrement("langfuse.ingestion.lookup.hit", 1, {
         store: "clickhouse",
@@ -198,7 +183,6 @@ export class IngestionService {
     const finalScoreRecord: ScoreRecordInsertType =
       await this.mergeScoreRecords({
         clickhouseScoreRecord,
-        postgresScoreRecord,
         scoreRecords,
       });
 
@@ -231,31 +215,18 @@ export class IngestionService {
       minTimestamp === Infinity
         ? undefined
         : convertDateToClickhouseDateTime(new Date(minTimestamp));
-    const [postgresTraceRecord, clickhouseTraceRecord] = await Promise.all([
-      this.getPostgresRecord({
-        projectId,
-        entityId,
-        table: TableName.Traces,
-      }),
-      this.getClickhouseRecord({
-        projectId,
-        entityId,
-        table: TableName.Traces,
-        additionalFilters: {
-          whereCondition: timestamp
-            ? " AND timestamp >= {timestamp: DateTime64(3)} "
-            : "",
-          params: { timestamp },
-        },
-      }),
-    ]);
+    const clickhouseTraceRecord = await this.getClickhouseRecord({
+      projectId,
+      entityId,
+      table: TableName.Traces,
+      additionalFilters: {
+        whereCondition: timestamp
+          ? " AND timestamp >= {timestamp: DateTime64(3)} "
+          : "",
+        params: { timestamp },
+      },
+    });
 
-    if (postgresTraceRecord) {
-      recordIncrement("langfuse.ingestion.lookup.hit", 1, {
-        store: "postgres",
-        object: "trace",
-      });
-    }
     if (clickhouseTraceRecord) {
       recordIncrement("langfuse.ingestion.lookup.hit", 1, {
         store: "clickhouse",
@@ -265,7 +236,6 @@ export class IngestionService {
 
     const finalTraceRecord = await this.mergeTraceRecords({
       clickhouseTraceRecord,
-      postgresTraceRecord,
       traceRecords,
     });
 
@@ -340,34 +310,22 @@ export class IngestionService {
         ? undefined
         : convertDateToClickhouseDateTime(new Date(minStartTime));
 
-    const [postgresObservationRecord, clickhouseObservationRecord, prompt] =
-      await Promise.all([
-        this.getPostgresRecord({
-          projectId,
-          entityId,
-          table: TableName.Observations,
-        }),
-        this.getClickhouseRecord({
-          projectId,
-          entityId,
-          table: TableName.Observations,
-          additionalFilters: {
-            whereCondition: `AND type = {type: String} ${startTime ? "AND start_time >= {startTime: DateTime64(3)} " : ""}`,
-            params: {
-              type,
-              startTime,
-            },
+    const [clickhouseObservationRecord, prompt] = await Promise.all([
+      this.getClickhouseRecord({
+        projectId,
+        entityId,
+        table: TableName.Observations,
+        additionalFilters: {
+          whereCondition: `AND type = {type: String} ${startTime ? "AND start_time >= {startTime: DateTime64(3)} " : ""}`,
+          params: {
+            type,
+            startTime,
           },
-        }),
-        this.getPrompt(projectId, observationEventList),
-      ]);
+        },
+      }),
+      this.getPrompt(projectId, observationEventList),
+    ]);
 
-    if (postgresObservationRecord) {
-      recordIncrement("langfuse.ingestion.lookup.hit", 1, {
-        store: "postgres",
-        object: "observation",
-      });
-    }
     if (clickhouseObservationRecord) {
       recordIncrement("langfuse.ingestion.lookup.hit", 1, {
         store: "clickhouse",
@@ -385,7 +343,6 @@ export class IngestionService {
     const finalObservationRecord = await this.mergeObservationRecords({
       projectId,
       observationRecords,
-      postgresObservationRecord,
       clickhouseObservationRecord,
     });
 
@@ -417,17 +374,14 @@ export class IngestionService {
 
   private async mergeScoreRecords(params: {
     scoreRecords: ScoreRecordInsertType[];
-    postgresScoreRecord?: ScoreRecordInsertType | null;
     clickhouseScoreRecord?: ScoreRecordInsertType | null;
   }): Promise<ScoreRecordInsertType> {
-    const { scoreRecords, postgresScoreRecord, clickhouseScoreRecord } = params;
+    const { scoreRecords, clickhouseScoreRecord } = params;
 
     // Set clickhouse first as this is the baseline for immutable fields
-    const recordsToMerge = [
-      clickhouseScoreRecord,
-      postgresScoreRecord,
-      ...scoreRecords,
-    ].filter(Boolean) as ScoreRecordInsertType[];
+    const recordsToMerge = [clickhouseScoreRecord, ...scoreRecords].filter(
+      Boolean,
+    ) as ScoreRecordInsertType[];
 
     const mergedRecord = this.mergeRecords(
       recordsToMerge,
@@ -439,17 +393,14 @@ export class IngestionService {
 
   private async mergeTraceRecords(params: {
     traceRecords: TraceRecordInsertType[];
-    postgresTraceRecord?: TraceRecordInsertType | null;
     clickhouseTraceRecord?: TraceRecordInsertType | null;
   }): Promise<TraceRecordInsertType> {
-    const { traceRecords, postgresTraceRecord, clickhouseTraceRecord } = params;
+    const { traceRecords, clickhouseTraceRecord } = params;
 
     // Set clickhouse first as this is the baseline for immutable fields
-    const recordsToMerge = [
-      clickhouseTraceRecord,
-      postgresTraceRecord,
-      ...traceRecords,
-    ].filter(Boolean) as TraceRecordInsertType[];
+    const recordsToMerge = [clickhouseTraceRecord, ...traceRecords].filter(
+      Boolean,
+    ) as TraceRecordInsertType[];
 
     const mergedRecord = this.mergeRecords(
       recordsToMerge,
@@ -462,20 +413,14 @@ export class IngestionService {
   private async mergeObservationRecords(params: {
     projectId: string;
     observationRecords: ObservationRecordInsertType[];
-    postgresObservationRecord?: ObservationRecordInsertType | null;
     clickhouseObservationRecord?: ObservationRecordInsertType | null;
   }): Promise<ObservationRecordInsertType> {
-    const {
-      projectId,
-      observationRecords,
-      postgresObservationRecord,
-      clickhouseObservationRecord,
-    } = params;
+    const { projectId, observationRecords, clickhouseObservationRecord } =
+      params;
 
     // Set clickhouse first as this is the baseline for immutable fields
     const recordsToMerge = [
       clickhouseObservationRecord,
-      postgresObservationRecord,
       ...observationRecords,
     ].filter(Boolean) as ObservationRecordInsertType[];
 
@@ -500,14 +445,6 @@ export class IngestionService {
       observationRecord: parsedObservationRecord,
     });
 
-    if (
-      "usage_details" in generationUsage &&
-      Object.keys(generationUsage.usage_details).length === 0
-    ) {
-      generationUsage.usage_details =
-        postgresObservationRecord?.usage_details ?? {};
-    }
-
     return {
       ...parsedObservationRecord,
       ...generationUsage,
@@ -844,58 +781,6 @@ export class IngestionService {
     );
   }
 
-  private async getPostgresRecord(params: {
-    projectId: string;
-    entityId: string;
-    table: TableName.Traces;
-  }): Promise<TraceRecordInsertType | null>;
-  private async getPostgresRecord(params: {
-    projectId: string;
-    entityId: string;
-    table: TableName.Scores;
-  }): Promise<ScoreRecordInsertType | null>;
-  private async getPostgresRecord(params: {
-    projectId: string;
-    entityId: string;
-    table: TableName.Observations;
-  }): Promise<ObservationRecordInsertType | null>;
-  private async getPostgresRecord(params: {
-    projectId: string;
-    entityId: string;
-    table: TableName;
-  }) {
-    const recordParser = {
-      traces: convertPostgresTraceToInsert,
-      scores: convertPostgresScoreToInsert,
-      observations: convertPostgresObservationToInsert,
-    };
-    const { projectId, entityId, table } = params;
-
-    const query =
-      table === TableName.Observations
-        ? Prisma.sql`
-          SELECT o.*,
-                 o."modelParameters" as model_parameters,
-                 p.name as prompt_name,
-                 p.version as prompt_version
-          FROM observations o
-          LEFT JOIN prompts p ON o.prompt_id = p.id
-          WHERE o.project_id = ${projectId}
-          AND o.id = ${entityId}
-          LIMIT 1;`
-        : Prisma.sql`
-          SELECT *
-          FROM ${Prisma.raw(table)}
-          WHERE project_id = ${projectId}
-          AND id = ${entityId}
-          LIMIT 1;`;
-
-    const result =
-      await this.prisma.$queryRaw<Array<Record<string, unknown>>>(query);
-
-    return result.length === 0 ? null : recordParser[table](result[0]);
-  }
-
   private mapTraceEventsToRecords(params: {
     traceEventList: TraceEventType[];
     projectId: string;

worker/src/services/IngestionService/tests/IngestionService.integration.test.ts

@@ -1202,68 +1202,6 @@ describe("Ingestion end-to-end tests", () => {
     expect(score.project_id).toBe("7a88fb47-b4e2-43b8-a06c-a5ce950dc53a");
   });
 
-  it("should merge observations from postgres and event list", async () => {
-    const traceId = randomUUID();
-    const observationId = randomUUID();
-
-    const latestEvent = new Date();
-    const oldEvent = new Date(latestEvent).setSeconds(
-      latestEvent.getSeconds() - 1,
-    );
-
-    await prisma.observation.create({
-      data: {
-        id: observationId,
-        type: "GENERATION",
-        traceId,
-        name: "generation-name",
-        input: { key: "value" },
-        output: "should be overwritten",
-        model: "gpt-3.5",
-        projectId,
-        startTime: new Date(oldEvent),
-        completionTokens: 5,
-        // Validates that numbers are parsed correctly. Since there is no usage, no effect on result
-        calculatedTotalCost: "0.273330000000000000000000000000",
-        modelParameters: { hello: "world" },
-      },
-    });
-
-    const observationEventList: ObservationEvent[] = [
-      {
-        id: randomUUID(),
-        type: "generation-create",
-        timestamp: new Date().toISOString(),
-        body: {
-          id: observationId,
-          traceId: traceId,
-          output: "overwritten",
-          usage: undefined,
-        },
-      },
-    ];
-
-    await ingestionService.processObservationEventList({
-      projectId,
-      entityId: observationId,
-      observationEventList,
-    });
-
-    await clickhouseWriter.flushAll(true);
-
-    const observation = await getClickhouseRecord(
-      TableName.Observations,
-      observationId,
-    );
-
-    expect(observation.name).toBe("generation-name");
-    expect(observation.input).toBe(JSON.stringify({ key: "value" }));
-    expect(observation.output).toBe("overwritten");
-    expect(observation.model_parameters).toBe('{"hello":"world"}');
-    expect(observation.usage_details.output).toBe(5);
-    expect(observation.project_id).toBe("7a88fb47-b4e2-43b8-a06c-a5ce950dc53a");
-  });
-
   it("should merge observations and set negative tokens and cost to null", async () => {
     await prisma.model.create({
       data: {