Ipc compression option

materializationengine/blueprints/client/api2.py

@@ -183,6 +183,20 @@ def __schema__(self):
  that are not valid at the timestamp that is queried. If True the filter will likely \
 not be relevant and the user might not be getting data back that they expect, but it will not error.",
 )
+query_parser.add_argument(
+    "ipc_compress",
+    type=inputs.boolean,
+    default=True,
+    required=False,
+    location="args",
+    help="whether to have arrow compress the result when using \
+          return_pyarrow=True and arrow_format=True. \
+          If False, the result will not have it's internal data\
+          compressed (note that the entire response \
+          will be gzip compressed if accept-enconding includes gzip). \
+          If True, accept-encoding will determine what \
+          internal compression is used",
+)
 
 
 metadata_parser = reqparse.RequestParser()
@@ -1187,6 +1201,7 @@ def post(self, datastack_name: str):
             desired_resolution=user_data["desired_resolution"],
             return_pyarrow=args["return_pyarrow"],
             arrow_format=args["arrow_format"],
+            ipc_compress=args["ipc_compress"],
         )
 
 
@@ -1422,6 +1437,7 @@ def post(
             desired_resolution=data["desired_resolution"],
             return_pyarrow=args["return_pyarrow"],
             arrow_format=args["arrow_format"],
+            ipc_compress=args["ipc_compress"],
         )
 
 

materializationengine/blueprints/client/common.py

@@ -281,6 +281,7 @@ def handle_simple_query(
         desired_resolution=data["desired_resolution"],
         return_pyarrow=args["return_pyarrow"],
         arrow_format=args["arrow_format"],
+        ipc_compress=args["ipc_compress"],
     )
 
 
@@ -293,7 +294,6 @@ def handle_complex_query(
     data,
     convert_desired_resolution=False,
 ):
-
     aligned_volume_name, pcg_table_name = get_relevant_datastack_info(datastack_name)
     db = dynamic_annotation_cache.get_db(aligned_volume_name)
 
@@ -447,4 +447,5 @@ def handle_complex_query(
         desired_resolution=data["desired_resolution"],
         return_pyarrow=args["return_pyarrow"],
         arrow_format=args["arrow_format"],
+        ipc_compress=args["ipc_compress"],
     )

materializationengine/blueprints/client/utils.py

@@ -5,7 +5,6 @@
 
 
 def collect_crud_columns(column_names):
-
     crud_columns = []
     created_columns = []
     for table in column_names.keys():
@@ -20,7 +19,6 @@ def collect_crud_columns(column_names):
 
 
 def after_request(response):
-
     accept_encoding = request.headers.get("Accept-Encoding", "")
 
     if "gzip" not in accept_encoding.lower():
@@ -68,6 +66,7 @@ def create_query_response(
     column_names,
     return_pyarrow=True,
     arrow_format=False,
+    ipc_compress=True,
 ):
     accept_encoding = request.headers.get("Accept-Encoding", "")
 
@@ -79,10 +78,13 @@ def create_query_response(
         if arrow_format:
             batch = pa.RecordBatch.from_pandas(df)
             sink = pa.BufferOutputStream()
-            if "lz4" in accept_encoding:
-                compression = "LZ4_FRAME"
-            elif "zstd" in accept_encoding:
-                compression = "ZSTD"
+            if ipc_compress:
+                if "lz4" in accept_encoding:
+                    compression = "LZ4_FRAME"
+                elif "zstd" in accept_encoding:
+                    compression = "ZSTD"
+                else:
+                    compression = None
             else:
                 compression = None
             opt = pa.ipc.IpcWriteOptions(compression=compression)