feat: support insert trace through the distributed table

lib/db/maintain/scripts.js

@@ -278,7 +278,33 @@ module.exports.traces_dist = [
   span_id FixedString(8),
   timestamp_ns Int64,
   duration Int64
-) ENGINE = Distributed('{{CLUSTER}}','{{DB}}', 'tempo_traces_attrs_gin', sipHash64(oid, trace_id)) {{{DIST_CREATE_SETTINGS}}};`
+) ENGINE = Distributed('{{CLUSTER}}','{{DB}}', 'tempo_traces_attrs_gin', sipHash64(oid, trace_id)) {{{DIST_CREATE_SETTINGS}}};`,
+
+  `CREATE TABLE IF NOT EXISTS {{DB}}.traces_input_dist {{{OnCluster}}} AS {{DB}}.traces_input;`,
+
+  `CREATE MATERIALIZED VIEW IF NOT EXISTS {{DB}}.traces_input_traces_dist_mv {{{OnCluster}}} TO tempo_traces_dist AS
+    SELECT  oid, 
+      unhex(trace_id)::FixedString(16) as trace_id,
+      unhex(span_id)::FixedString(8) as span_id,
+      unhex(parent_id) as parent_id,
+      name,
+      timestamp_ns,
+      duration_ns,
+      service_name,
+      payload_type,
+      payload
+    FROM traces_input_dist`,
+
+  `CREATE MATERIALIZED VIEW IF NOT EXISTS {{DB}}.traces_input_tags_dist_mv {{{OnCluster}}} TO tempo_traces_attrs_gin_dist AS
+    SELECT  oid,
+      toDate(intDiv(timestamp_ns, 1000000000)) as date,
+      tags.1 as key, 
+      tags.2 as val,
+      unhex(trace_id)::FixedString(16) as trace_id, 
+      unhex(span_id)::FixedString(8) as span_id, 
+      timestamp_ns,      
+      duration_ns as duration
+    FROM traces_input_dist ARRAY JOIN tags`,
 ]
 
 module.exports.profiles = [