diff --git a/query-service-impl/build.gradle.kts b/query-service-impl/build.gradle.kts
index 7b1d0e2b..d227935a 100644
--- a/query-service-impl/build.gradle.kts
+++ b/query-service-impl/build.gradle.kts
@@ -34,6 +34,9 @@ dependencies {
     implementation("org.apache.helix:helix-core:1.3.0") {
       because("CVE-2022-47500")
     }
+    implementation("org.apache.zookeeper:zookeeper:3.7.2") {
+      because("CVE-2023-44981")
+    }
     implementation("org.webjars:swagger-ui:5.1.0") {
       because("CVE-2019-16728,CVE-2020-26870")
     }
diff --git a/query-service-impl/src/main/java/org/hypertrace/core/query/service/trino/converters/DefaultColumnRequestConverter.java b/query-service-impl/src/main/java/org/hypertrace/core/query/service/trino/converters/DefaultColumnRequestConverter.java
index 561a79c3..b30ef3a5 100644
--- a/query-service-impl/src/main/java/org/hypertrace/core/query/service/trino/converters/DefaultColumnRequestConverter.java
+++ b/query-service-impl/src/main/java/org/hypertrace/core/query/service/trino/converters/DefaultColumnRequestConverter.java
@@ -243,37 +243,60 @@ private void handleConversionForArrayColumnExpression(
 
     if (handleConversionForNullOrEmptyArrayLiteral(lhs, operator, builder, value)) return;
 
-    // support only equals and IN operator
-    // both of them are handled as contains check to align with existing implementation
-    if (operator.equals("=")
-        || operator.equals("IN")
-        || operator.equals("!=")
-        || operator.equals("NOT IN")) {
-      // add NOT operator to negate the match condition
-      if (operator.equals("!=") || operator.equals("NOT IN")) {
+    // Support only equals and IN operator
+    if (operator.equals("=") || operator.equals("!=")) {
+      // Equals (=): CONTAINS(ip_types, 'Bot')
+      // Not equals (!=): NOT CONTAINS(ip_types, 'Bot')
+      if (operator.equals("!=")) {
         builder.append("NOT ");
       }
+      builder.append("CONTAINS(");
       builder.append(lhs);
-      // overlap operator for array
-      builder.append(" && ");
+      builder.append(", ");
+      if (value.getValueType() == ValueType.STRING) {
+        builder.append(QUESTION_MARK);
+        paramsBuilder.addStringParam(value.getString());
+      } else {
+        throw new IllegalArgumentException(
+            String.format("Unsupported value {%s} for array column", value));
+      }
+      builder.append(")");
+    } else if (operator.equals("IN") || operator.equals("NOT IN")) {
+      // IN: CARDINALITY(ARRAY_INTERSECT(ip_types, ARRAY['Public Proxy', 'Bot'])) > 0
+      // NOT IN: CARDINALITY(ARRAY_INTERSECT(ip_types, ARRAY['Public Proxy', 'Bot'])) = 0
+      builder.append("CARDINALITY(");
+      builder.append("ARRAY_INTERSECT(");
+      builder.append(lhs);
+      builder.append(", ARRAY[");
+      switch (value.getValueType()) {
+        case STRING:
+          builder.append(QUESTION_MARK);
+          paramsBuilder.addStringParam(value.getString());
+          break;
+        case STRING_ARRAY:
+          String delim = "";
+          for (String item : value.getStringArrayList()) {
+            builder.append(delim);
+            builder.append(QUESTION_MARK);
+            paramsBuilder.addStringParam(item);
+            delim = ", ";
+          }
+          break;
+        default:
+          throw new IllegalArgumentException(
+              String.format("Unsupported value {%s} for array column", value));
+      }
+      builder.append("]))");
+      if (operator.equals("IN")) {
+        builder.append(" > 0");
+      } else {
+        builder.append(" = 0");
+      }
     } else {
       throw new IllegalArgumentException(
           String.format(
               "Unsupported operator {%s} for array column with non-empty value", operator));
     }
-    builder.append("?");
-    switch (value.getValueType()) {
-      case STRING:
-        paramsBuilder.addStringParam("{" + value.getString() + "}");
-        break;
-      case STRING_ARRAY:
-        paramsBuilder.addStringParam("{" + String.join(", ", value.getStringArrayList()) + "}");
-        break;
-      default:
-        throw new IllegalArgumentException(
-            String.format("Unsupported value {%s} for array column", value));
-    }
-    builder.append(QUESTION_MARK);
   }
 
   private boolean handleConversionForNullOrEmptyArrayLiteral(
diff --git a/query-service-impl/src/test/java/org/hypertrace/core/query/service/trino/QueryRequestToTrinoSQLConverterTest.java b/query-service-impl/src/test/java/org/hypertrace/core/query/service/trino/QueryRequestToTrinoSQLConverterTest.java
index b84d7bca..5521a9fd 100644
--- a/query-service-impl/src/test/java/org/hypertrace/core/query/service/trino/QueryRequestToTrinoSQLConverterTest.java
+++ b/query-service-impl/src/test/java/org/hypertrace/core/query/service/trino/QueryRequestToTrinoSQLConverterTest.java
@@ -175,7 +175,7 @@ void testQueryWithStringFilter() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testSQLiWithStringValueFilter() {
     QueryRequest queryRequest =
         buildSimpleQueryWithFilter(
@@ -185,7 +185,7 @@ void testSQLiWithStringValueFilter() {
 
     assertSQLQuery(
         queryRequest,
-        "Select encode(span_id, 'hex') FROM public.\"span-event-view\" WHERE "
+        "Select lower(to_hex(span_id)) FROM span-event-view WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
@@ -316,7 +316,7 @@ void testQueryWithDoubleFilter() {
 
     assertSQLQuery(
         queryRequest,
-        "Select encode(span_id, 'hex') FROM public.\"span-event-view\" WHERE "
+        "Select lower(to_hex(span_id)) FROM span-event-view WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
@@ -384,14 +384,14 @@ void testQueryWithTimestampFilter() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithOrderBy() {
     TableDefinition tableDefinition = getDefaultTableDefinition();
     defaultMockingForExecutionContext();
 
     assertSQLQuery(
         buildOrderByQuery(),
-        "Select encode(span_id, 'hex'), start_time_millis, end_time_millis FROM public.\"span-event-view\" WHERE "
+        "Select lower(to_hex(span_id)), start_time_millis, end_time_millis FROM span-event-view WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
@@ -401,7 +401,7 @@ void testQueryWithOrderBy() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithOrderByWithPagination() {
     QueryRequest orderByQueryRequest = buildOrderByQuery();
     Builder builder = QueryRequest.newBuilder(orderByQueryRequest);
@@ -411,7 +411,7 @@ void testQueryWithOrderByWithPagination() {
 
     assertSQLQuery(
         builder.build(),
-        "Select encode(span_id, 'hex'), start_time_millis, end_time_millis FROM public.\"span-event-view\" WHERE "
+        "Select lower(to_hex(span_id)), start_time_millis, end_time_millis FROM span-event-view WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
@@ -421,7 +421,7 @@ void testQueryWithOrderByWithPagination() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithGroupByWithMultipleAggregates() {
     QueryRequest groupByQueryRequest = buildMultipleGroupByMultipleAggQuery();
     Builder builder = QueryRequest.newBuilder(groupByQueryRequest);
@@ -431,7 +431,7 @@ void testQueryWithGroupByWithMultipleAggregates() {
 
     assertSQLQuery(
         builder.build(),
-        "select count(*), avg(duration_millis) FROM public.\"span-event-view\""
+        "select count(*), avg(duration_millis) FROM span-event-view"
             + " where "
             + tableDefinition.getTenantIdColumn()
             + " = '"
@@ -443,7 +443,7 @@ void testQueryWithGroupByWithMultipleAggregates() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithGroupByWithMultipleAggregatesAndOrderBy() {
     QueryRequest orderByQueryRequest = buildMultipleGroupByMultipleAggAndOrderByQuery();
     Builder builder = QueryRequest.newBuilder(orderByQueryRequest);
@@ -453,7 +453,7 @@ void testQueryWithGroupByWithMultipleAggregatesAndOrderBy() {
 
     assertSQLQuery(
         builder.build(),
-        "select count(*), avg(duration_millis) FROM public.\"span-event-view\""
+        "select count(*), avg(duration_millis) FROM span-event-view"
             + " where "
             + tableDefinition.getTenantIdColumn()
             + " = '"
@@ -465,7 +465,7 @@ void testQueryWithGroupByWithMultipleAggregatesAndOrderBy() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithDistinctCountAggregation() {
     Filter startTimeFilter =
         createTimeFilter("Span.start_time_millis", Operator.GT, 1570658506605L);
@@ -489,7 +489,7 @@ void testQueryWithDistinctCountAggregation() {
 
     assertSQLQuery(
         queryRequest,
-        "select count(distinct encode(span_id, 'hex')) FROM public.\"span-event-view\""
+        "select count(distinct lower(to_hex(span_id))) FROM span-event-view"
             + " where "
             + tableDefinition.getTenantIdColumn()
             + " = '"
@@ -501,7 +501,7 @@ void testQueryWithDistinctCountAggregation() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithDistinctCountAggregationAndGroupBy() {
     Filter startTimeFilter =
         createTimeFilter("Span.start_time_millis", Operator.GT, 1570658506605L);
@@ -532,14 +532,14 @@ void testQueryWithDistinctCountAggregationAndGroupBy() {
 
     assertSQLQuery(
         queryRequest,
-        "select encode(span_id, 'hex'), count(distinct encode(span_id, 'hex')) FROM public.\"span-event-view\""
+        "select lower(to_hex(span_id)), count(distinct lower(to_hex(span_id))) FROM span-event-view"
             + " where "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
             + "' "
             + "and ( start_time_millis > 1570658506605 and end_time_millis < 1570744906673 )"
-            + " group by span_id order by count(distinct span_id) limit 15",
+            + " group by 1 order by 2 limit 15",
         tableDefinition,
         executionContext);
   }
@@ -575,7 +575,7 @@ void testQueryWithStringArray() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testSQLiWithStringArrayFilter() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.displaySpanName"));
@@ -590,7 +590,7 @@ void testSQLiWithStringArrayFilter() {
 
     assertSQLQuery(
         builder.build(),
-        "SELECT span_name FROM public.\"span-event-view\" WHERE "
+        "SELECT span_name FROM span-event-view WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
@@ -895,12 +895,12 @@ void testQueryWithBytesColumnInFilter() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithArrayColumnWithValidValue() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.id").build());
 
-    Filter filter = createEqualsFilter("Span.labels", "label1");
+    Filter filter = createEqualsFilter("Span.ip_types", "Bot");
     builder.setFilter(filter);
     builder.setLimit(5);
 
@@ -910,13 +910,13 @@ void testQueryWithArrayColumnWithValidValue() {
 
     assertSQLQuery(
         request,
-        "SELECT encode(span_id, 'hex') FROM public.\"span-event-view\" "
+        "SELECT lower(to_hex(span_id)) FROM span-event-view "
             + "WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
             + "' "
-            + "AND labels && '{label1}' limit 5",
+            + "AND CONTAINS(ip_types, 'Bot') limit 5",
         tableDefinition,
         executionContext);
   }
@@ -973,12 +973,12 @@ void testQueryWithArrayColumnWithEmptyValue() {
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithArrayColumnNotEqualsFilter() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.id").build());
 
-    Filter filter = createNotEqualsFilter("Span.labels", "label1");
+    Filter filter = createNotEqualsFilter("Span.ip_types", "Bot");
     builder.setFilter(filter);
     builder.setLimit(5);
 
@@ -988,23 +988,23 @@ void testQueryWithArrayColumnNotEqualsFilter() {
 
     assertSQLQuery(
         request,
-        "SELECT encode(span_id, 'hex') FROM public.\"span-event-view\" "
+        "SELECT lower(to_hex(span_id)) FROM span-event-view "
             + "WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
             + "' "
-            + "AND NOT labels && '{label1}' limit 5",
+            + "AND NOT CONTAINS(ip_types, 'Bot') limit 5",
         tableDefinition,
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithArrayColumnInFilter() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.id").build());
 
-    Filter filter = createInFilter("Span.labels", List.of("label1", "label2"));
+    Filter filter = createInFilter("Span.ip_types", List.of("Public Proxy", "Bot"));
     builder.setFilter(filter);
     builder.setLimit(5);
 
@@ -1014,23 +1014,23 @@ void testQueryWithArrayColumnInFilter() {
 
     assertSQLQuery(
         request,
-        "SELECT encode(span_id, 'hex') FROM public.\"span-event-view\" "
+        "SELECT lower(to_hex(span_id)) FROM span-event-view "
             + "WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
             + "' "
-            + "AND labels && '{label1, label2}' limit 5",
+            + "AND CARDINALITY(ARRAY_INTERSECT(ip_types, ARRAY['Public Proxy', 'Bot'])) > 0 limit 5",
         tableDefinition,
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithArrayColumnNotInFilter() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.id").build());
 
-    Filter filter = createNotInFilter("Span.labels", List.of("label1", "label2"));
+    Filter filter = createNotInFilter("Span.ip_types", List.of("Public Proxy", "Bot"));
     builder.setFilter(filter);
     builder.setLimit(5);
 
@@ -1040,18 +1040,18 @@ void testQueryWithArrayColumnNotInFilter() {
 
     assertSQLQuery(
         request,
-        "SELECT encode(span_id, 'hex') FROM public.\"span-event-view\" "
+        "SELECT lower(to_hex(span_id)) FROM span-event-view "
             + "WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
             + "' "
-            + "AND NOT labels && '{label1, label2}' limit 5",
+            + "AND CARDINALITY(ARRAY_INTERSECT(ip_types, ARRAY['Public Proxy', 'Bot'])) = 0 limit 5",
         tableDefinition,
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithStringColumnWithNullString() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.id"));
@@ -1068,18 +1068,18 @@ void testQueryWithStringColumnWithNullString() {
 
     assertSQLQuery(
         request,
-        "SELECT encode(span_id, 'hex') FROM public.\"span-event-view\" "
+        "SELECT lower(to_hex(span_id)) FROM span-event-view "
             + "WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
             + TENANT_ID
             + "' "
-            + "AND ( span_id != '' ) limit 5",
+            + "AND ( lower(to_hex(span_id)) != '' ) limit 5",
         tableDefinition,
         executionContext);
   }
 
-  // @Test
+  @Test
   void testQueryWithLongColumn() {
     Builder builder = QueryRequest.newBuilder();
     builder.addSelection(createColumnExpression("Span.id"));
@@ -1100,7 +1100,7 @@ void testQueryWithLongColumn() {
 
     assertSQLQuery(
         request,
-        "SELECT encode(span_id, 'hex') FROM public.\"span-event-view\" "
+        "SELECT lower(to_hex(span_id)) FROM span-event-view "
             + "WHERE "
             + tableDefinition.getTenantIdColumn()
             + " = '"
@@ -1432,7 +1432,7 @@ private QueryRequest buildMultipleGroupByMultipleAggAndOrderByQuery() {
             SortOrder.DESC));
     builder.addOrderBy(
         createOrderByExpression(
-            createAliasedFunctionExpression("COUNT", "Span.id", "count_encode(span_id, 'hex')"),
+            createAliasedFunctionExpression("COUNT", "Span.id", "count_lower(to_hex(span_id))"),
             SortOrder.DESC));
     return builder.build();
   }
diff --git a/query-service-impl/src/test/resources/trino_request_handler.conf b/query-service-impl/src/test/resources/trino_request_handler.conf
index 146edf65..111b1c9f 100644
--- a/query-service-impl/src/test/resources/trino_request_handler.conf
+++ b/query-service-impl/src/test/resources/trino_request_handler.conf
@@ -7,7 +7,7 @@
       tableName = span-event-view
       mapFields = ["tags", "request_headers"]
       bytesFields = ["parent_span_id", "span_id"]
-      arrayFields = ["labels"]
+      arrayFields = ["ip_types"]
       tdigestFields = ["response_time_millis_tdigest"]
       fieldMap = {
         "Span.tags": "tags",
@@ -28,7 +28,7 @@
         "Span.metrics.duration_millis": "duration_millis",
         "Span.serviceName": "service_name",
         "Span.attributes.parent_span_id": "parent_span_id",
-        "Span.labels": "labels",
+        "Span.ip_types": "ip_types",
         "Span.user_latitude": "user_latitude",
       }
     }