Repository: Bears-R-Us/arkouda
Issue category: Performance Issue

Issue:
Multi-column Parquet write inefficiencies
Passing along a comment from another developer. " I've noticed that Parquet output from Arkouoda is currently very slow. The Arkouda client library saves multi-column dataframes by sending a separate writeParquet request to the server for each column, adding them to the file one-by-one. The ccp_appendColumnToParquet function doesn't add the column to the file in-place, though; instead it loads the whole existing file into memory, adds the new column to the table object in memory, and overwrites the whole file with the new version. That means it's loading and re-saving the content of all the old columns each time it adds a new one, which adds a lot of overhead since this happens for every column."
Multi-column Parquet write inefficiencies
Passing along a comment from another developer. " I've noticed that Parquet output from Arkouoda is currently very slow. The Arkouda client library saves multi-column dataframes by sending a separate writeParquet request to the server for each column, adding them to the file one-by-one. The ccp_appendColumnToParquet function doesn't add the column to the file in-place, though; instead it loads the whole existing file into memory, adds the new column to the table object in memory, and overwrites the whole file with the new version. That means it's loading and re-saving the content of all the old columns each time it adds a new one, which adds a lot of overhead since this happens for every column."


Hints:
Thanks for bringing this issue to our attention. When the Parquet work was implemented, it was before the dataframe functionality had been implemented, so that use case wasn't considered in the initial design. As you have seen, the Parquet append functionality is very sub-optimal and that is because Parquet does not support appending to files natively, so we had to work around that by reading in the whole file and then rewriting it.

I do not have a great deal of familiarity with the dataframe code as it exists today, but I suspect that we should be able to implement a much faster implementation that only writes to the file once. I hope to get to this in the near future.
Thank you!
Spoke with @bmcdonald3. To allow me to get more familiar with the Parquet code, I am going to take this one. 
There are some challenges presented by this that I will need to look into the best way to handle. Currently, all our Parquet support is designed to operate on a single Chapel array. In this case, we will most likely have instances where we have multiple Chapel Arrays of varying types needing to be written. In HDF5, this is a bit easier because we operate on the Objects directly in Chapel without needing to format them. Additionally, because the Apache Arrow library is C++ only, we have to make calls from Chapel to C and from C to C++. This should not be an issue, but noting as an additional complexity. 

Ideally, I would like to configure methods to interact with objects based on ids/pointers like we are able to with HDF5. I am looking into configuring this now.
Thanks for bringing this issue to our attention. When the Parquet work was implemented, it was before the dataframe functionality had been implemented, so that use case wasn't considered in the initial design. As you have seen, the Parquet append functionality is very sub-optimal and that is because Parquet does not support appending to files natively, so we had to work around that by reading in the whole file and then rewriting it.

I do not have a great deal of familiarity with the dataframe code as it exists today, but I suspect that we should be able to implement a much faster implementation that only writes to the file once. I hope to get to this in the near future.
Thank you!
Spoke with @bmcdonald3. To allow me to get more familiar with the Parquet code, I am going to take this one. 
There are some challenges presented by this that I will need to look into the best way to handle. Currently, all our Parquet support is designed to operate on a single Chapel array. In this case, we will most likely have instances where we have multiple Chapel Arrays of varying types needing to be written. In HDF5, this is a bit easier because we operate on the Objects directly in Chapel without needing to format them. Additionally, because the Apache Arrow library is C++ only, we have to make calls from Chapel to C and from C to C++. This should not be an issue, but noting as an additional complexity. 

Ideally, I would like to configure methods to interact with objects based on ids/pointers like we are able to with HDF5. I am looking into configuring this now.

Patch:
diff --git a/arkouda/dataframe.py b/arkouda/dataframe.py
index 10c98b1c69..6291b22ff8 100644
--- a/arkouda/dataframe.py
+++ b/arkouda/dataframe.py
@@ -1457,7 +1457,7 @@ def _prep_data(self, index=False, columns=None):
             data = {c: self.data[c] for c in columns}
 
         if index:
-            data["Index"] = self.index
+            data["Index"] = self.index.values
         return data
 
     def to_hdf(self, path, index=False, columns=None, file_type="distribute"):
diff --git a/arkouda/io.py b/arkouda/io.py
index f386ac3a77..cdfbd6b3b2 100644
--- a/arkouda/io.py
+++ b/arkouda/io.py
@@ -1,7 +1,6 @@
 import glob
 import json
 import os
-import warnings
 from typing import Dict, List, Mapping, Optional, Union, cast
 from warnings import warn
 
@@ -334,7 +333,7 @@ def _parse_errors(rep_msg, allow_errors: bool = False):
     file_errors = rep_msg["file_errors"] if "file_errors" in rep_msg else []
     if allow_errors and file_errors:
         file_error_count = rep_msg["file_error_count"] if "file_error_count" in rep_msg else -1
-        warnings.warn(
+        warn(
             f"There were {file_error_count} errors reading files on the server. "
             + f"Sample error messages {file_errors}",
             RuntimeWarning,
@@ -822,10 +821,8 @@ def to_parquet(
     mode : {'truncate' | 'append'}
         By default, truncate (overwrite) the output files if they exist.
         If 'append', attempt to create new dataset in existing files.
-    file_type : str ("single" | "distribute")
-            Default: distribute
-            Single writes the dataset to a single file
-            Distribute writes the dataset to a file per locale
+        'append' is deprecated, please use the multi-column write
+
 
     Returns
     -------
@@ -865,12 +862,32 @@ def to_parquet(
     if mode.lower() not in ["append", "truncate"]:
         raise ValueError("Allowed modes are 'truncate' and 'append'")
 
-    datasetNames, pdarrays = _bulk_write_prep(columns, names)
+    if mode.lower() == "append":
+        warn(
+            "Append has been deprecated when writing Parquet files. "
+            "Please write all columns to the file at once.",
+            DeprecationWarning,
+        )
 
-    for arr, name in zip(pdarrays, cast(List[str], datasetNames)):
-        arr.to_parquet(prefix_path=prefix_path, dataset=name, mode=mode, compressed=compressed)
-        if mode.lower() == "truncate":
-            mode = "append"
+    datasetNames, pdarrays = _bulk_write_prep(columns, names)
+    # append or single column use the old logic
+    if mode.lower() == "append" or len(pdarrays) == 1:
+        for arr, name in zip(pdarrays, cast(List[str], datasetNames)):
+            arr.to_parquet(prefix_path=prefix_path, dataset=name, mode=mode, compressed=compressed)
+    else:
+        print(cast(
+            str,
+            generic_msg(
+                cmd="toParquet_multi",
+                args={
+                    "columns": pdarrays,
+                    "col_names": datasetNames,
+                    "filename": prefix_path,
+                    "num_cols": len(pdarrays),
+                    "compressed": compressed,
+                },
+            )
+        ))
 
 
 def to_hdf(
@@ -1234,7 +1251,7 @@ def read(
     elif file_format == "parquet":
         cmd = "readAllParquet"
     else:
-        warnings.warn(f"Unrecognized file format string: {file_format}. Inferring file type")
+        warn(f"Unrecognized file format string: {file_format}. Inferring file type")
         cmd = "readany"
     if iterative:  # iterative calls to server readhdf
         return {
@@ -1266,7 +1283,7 @@ def read(
         file_errors = rep["file_errors"] if "file_errors" in rep else []
         if allow_errors and file_errors:
             file_error_count = rep["file_error_count"] if "file_error_count" in rep else -1
-            warnings.warn(
+            warn(
                 f"There were {file_error_count} errors reading files on the server. "
                 + f"Sample error messages {file_errors}",
                 RuntimeWarning,
diff --git a/src/ArrowFunctions.cpp b/src/ArrowFunctions.cpp
index cb6adfc0ec..74e9799be5 100644
--- a/src/ArrowFunctions.cpp
+++ b/src/ArrowFunctions.cpp
@@ -366,6 +366,128 @@ int cpp_readColumnByName(const char* filename, void* chpl_arr, const char* colna
   }
 }
 
+// configure the schema for a multicolumn file
+std::shared_ptr<parquet::schema::GroupNode> SetupSchema(void* column_names, void* datatypes, int64_t colnum) {
+  parquet::schema::NodeVector fields;
+  auto cname_ptr = (char**)column_names;
+  auto dtypes_ptr = (int64_t*) datatypes;
+  for (int64_t i = 0; i < colnum; i++){
+    if(dtypes_ptr[i] == ARROWINT64)
+      fields.push_back(parquet::schema::PrimitiveNode::Make(cname_ptr[i], parquet::Repetition::REQUIRED, parquet::Type::INT64, parquet::ConvertedType::NONE));
+    else if(dtypes_ptr[i] == ARROWUINT64)
+      fields.push_back(parquet::schema::PrimitiveNode::Make(cname_ptr[i], parquet::Repetition::REQUIRED, parquet::Type::INT64, parquet::ConvertedType::UINT_64));
+    else if(dtypes_ptr[i] == ARROWBOOLEAN)
+      fields.push_back(parquet::schema::PrimitiveNode::Make(cname_ptr[i], parquet::Repetition::REQUIRED, parquet::Type::BOOLEAN, parquet::ConvertedType::NONE));
+    else if(dtypes_ptr[i] == ARROWDOUBLE)
+      fields.push_back(parquet::schema::PrimitiveNode::Make(cname_ptr[i], parquet::Repetition::REQUIRED, parquet::Type::DOUBLE, parquet::ConvertedType::NONE));
+    else if(dtypes_ptr[i] == ARROWSTRING)
+      fields.push_back(parquet::schema::PrimitiveNode::Make(cname_ptr[i], parquet::Repetition::REQUIRED, parquet::Type::BYTE_ARRAY, parquet::ConvertedType::NONE));
+  }
+  return std::static_pointer_cast<parquet::schema::GroupNode>(
+      parquet::schema::GroupNode::Make("schema", parquet::Repetition::REQUIRED, fields));
+}
+
+int cpp_writeMultiColToParquet(const char* filename, void* column_names, 
+                                void** ptr_arr, void* datatypes,
+                                int64_t colnum, int64_t numelems, int64_t rowGroupSize,
+                                bool compressed, char** errMsg) {
+  try {
+    // initialize the file to write to
+    using FileClass = ::arrow::io::FileOutputStream;
+    std::shared_ptr<FileClass> out_file;
+    ARROWRESULT_OK(FileClass::Open(filename), out_file);
+
+    // Setup the parquet schema
+    std::shared_ptr<parquet::schema::GroupNode> schema = SetupSchema(column_names, datatypes, colnum);
+
+    parquet::WriterProperties::Builder builder;
+    if(compressed) {
+      builder.compression(parquet::Compression::SNAPPY);
+      builder.encoding(parquet::Encoding::RLE);
+    }
+    std::shared_ptr<parquet::WriterProperties> props = builder.build();
+
+    std::shared_ptr<parquet::ParquetFileWriter> file_writer =
+      parquet::ParquetFileWriter::Open(out_file, schema, props);
+
+    std::queue<int64_t> idxQueue; // queue used to track string byteIdx 
+
+    auto dtypes_ptr = (int64_t*) datatypes;
+    int64_t numLeft = numelems; // number of elements remaining to write (rows)
+    int64_t x = 0;  // index to start writing batch from
+    while (numLeft > 0) {
+      // Append a RowGroup with a specific number of rows.
+      parquet::RowGroupWriter* rg_writer = file_writer->AppendRowGroup();
+      int64_t batchSize = rowGroupSize;
+      if(numLeft < rowGroupSize)
+        batchSize = numLeft;
+
+      // loop the columns and write the row groups
+      for(int64_t i = 0; i < colnum; i++){
+        int64_t dtype = dtypes_ptr[i];
+        if (dtype == ARROWINT64 || dtype == ARROWUINT64) {
+          auto data_ptr = (int64_t*)ptr_arr[i];
+          parquet::Int64Writer* int64_writer =
+              static_cast<parquet::Int64Writer*>(rg_writer->NextColumn());
+          int64_writer->WriteBatch(batchSize, nullptr, nullptr, &data_ptr[x]);
+        } else if(dtype == ARROWBOOLEAN) {
+          auto data_ptr = (bool*)ptr_arr[i];
+          parquet::BoolWriter* bool_writer =
+            static_cast<parquet::BoolWriter*>(rg_writer->NextColumn());
+          bool_writer->WriteBatch(batchSize, nullptr, nullptr, &data_ptr[x]);
+        } else if(dtype == ARROWDOUBLE) {
+          auto data_ptr = (double*)ptr_arr[i];
+          parquet::DoubleWriter* dbl_writer =
+            static_cast<parquet::DoubleWriter*>(rg_writer->NextColumn());
+          dbl_writer->WriteBatch(batchSize, nullptr, nullptr, &data_ptr[x]);
+        } else if(dtype == ARROWSTRING) {
+          auto data_ptr = (uint8_t*)ptr_arr[i];
+          parquet::ByteArrayWriter* ba_writer =
+          static_cast<parquet::ByteArrayWriter*>(rg_writer->NextColumn());
+          int64_t count = 0;
+          int64_t byteIdx = 0;
+
+          // identify the starting byte index
+          if (x > 0){
+            byteIdx = idxQueue.front();
+            idxQueue.pop();
+          }
+          
+          while(count < batchSize) {
+            parquet::ByteArray value;
+            int16_t definition_level = 1;
+            value.ptr = reinterpret_cast<const uint8_t*>(&data_ptr[byteIdx]);
+            int64_t nextIdx = byteIdx;
+            while (data_ptr[nextIdx] != 0x00){
+              nextIdx++;
+            }
+            // subtract 1 since we have the null terminator
+            value.len = nextIdx - byteIdx;
+            ba_writer->WriteBatch(1, &definition_level, nullptr, &value);
+            count++;
+            byteIdx = nextIdx + 1;
+          }
+          if (numLeft - count > 0) {
+            idxQueue.push(byteIdx);
+          }
+        } else {
+          return ARROWERROR;
+        }
+      }
+      numLeft -= batchSize;
+      x += batchSize;
+    }
+
+    file_writer->Close();
+    ARROWSTATUS_OK(out_file->Close());
+    
+    return 0;
+   } catch (const std::exception& e) {
+    *errMsg = strdup(e.what());
+    return ARROWERROR;
+  }
+}
+
 int cpp_writeColumnToParquet(const char* filename, void* chpl_arr,
                              int64_t colnum, const char* dsetname, int64_t numelems,
                              int64_t rowGroupSize, int64_t dtype, bool compressed,
@@ -783,4 +905,11 @@ extern "C" {
   void c_free_string(void* ptr) {
     cpp_free_string(ptr);
   }
+
+  int c_writeMultiColToParquet(const char* filename, void* column_names, 
+                                void** ptr_arr, void* datatypes,
+                                int64_t colnum, int64_t numelems, int64_t rowGroupSize,
+                                bool compressed, char** errMsg){
+    return cpp_writeMultiColToParquet(filename, column_names, ptr_arr, datatypes, colnum, numelems, rowGroupSize, compressed, errMsg);
+  }
 }
diff --git a/src/ArrowFunctions.h b/src/ArrowFunctions.h
index 511e6fab11..e2902fd9f3 100644
--- a/src/ArrowFunctions.h
+++ b/src/ArrowFunctions.h
@@ -10,6 +10,7 @@
 #include <parquet/arrow/writer.h>
 #include <parquet/column_reader.h>
 #include <parquet/api/writer.h>
+#include <queue>
 extern "C" {
 #endif
 
@@ -79,6 +80,16 @@ extern "C" {
                                 const char* dsetname, int64_t numelems,
                                 int64_t dtype, bool compressed,
                                 char** errMsg);
+  
+  int c_writeMultiColToParquet(const char* filename, void* column_names, 
+                                void** ptr_arr, void* datatypes,
+                                int64_t colnum, int64_t numelems, int64_t rowGroupSize,
+                                bool compressed, char** errMsg);
+
+  int cpp_writeMultiColToParquet(const char* filename, void* column_names, 
+                                  void** ptr_arr, void* datatypes,
+                                  int64_t colnum, int64_t numelems, int64_t rowGroupSize,
+                                  bool compressed, char** errMsg);
     
   const char* c_getVersionInfo(void);
   const char* cpp_getVersionInfo(void);
diff --git a/src/ParquetMsg.chpl b/src/ParquetMsg.chpl
index f9521a8cf0..0730cb0677 100644
--- a/src/ParquetMsg.chpl
+++ b/src/ParquetMsg.chpl
@@ -12,6 +12,7 @@ module ParquetMsg {
   use NumPyDType;
   use Sort;
   use CommAggregation;
+  use AryUtil;
 
   use SegmentedString;
 
@@ -769,6 +770,245 @@ module ParquetMsg {
     }
   }
 
+  proc writeMultiColParquet(filename: string, col_names: [] string, 
+                              ncols: int, sym_names: [] string, targetLocales: [] locale, 
+                              compressed: bool, st: borrowed SymTab): bool throws {
+
+    extern proc c_writeMultiColToParquet(filename, column_names, ptr_arr,
+                                      datatypes, colnum, numelems, rowGroupSize, compressed, errMsg): int;
+
+    var prefix: string;
+    var extension: string;
+    (prefix, extension) = getFileMetadata(filename);
+
+    // Generate the filenames based upon the number of targetLocales.
+    var filenames = generateFilenames(prefix, extension, targetLocales.size);
+
+    //Generate a list of matching filenames to test against. 
+    var matchingFilenames = getMatchingFilenames(prefix, extension);
+
+    // TODO when APPEND is fully deprecated update this to not need the mode.
+    var filesExist = processParquetFilenames(filenames, matchingFilenames, TRUNCATE); // set to truncate. We will not be supporting appending. 
+
+    coforall (loc, idx) in zip(targetLocales, filenames.domain) do on loc {
+      var pqErr = new parquetErrorMsg();
+      const fname = filenames[idx];
+
+      var ptrList: [0..#ncols] c_void_ptr;
+      var datatypes: [0..#ncols] int;
+      var sizeList: [0..#ncols] int;
+
+      var my_column_names = col_names;
+      var c_names: [0..#ncols] c_string;
+
+      var locSize: int = 0;
+      var sections_sizes: [0..#ncols] int; // only fill in sizes for str columns
+      forall (i, column) in zip(0..#ncols, sym_names) with (+ reduce locSize) {
+        var entry = st.lookup(column);
+        // need to calculate the total size of Strings on this local
+        if (entry.isAssignableTo(SymbolEntryType.SegStringSymEntry)) {
+          var e: SegStringSymEntry = toSegStringSymEntry(entry);
+          var segStr = new SegString("", e);
+          ref ss = segStr;
+          var lens = ss.getLengths();
+          const locDom = ss.offsets.a.localSubdomain();
+          var x: int;
+          for i in locDom do x += lens[i];
+          sections_sizes[i] = x;
+          locSize += x;
+        }
+      }
+
+      var str_vals: [0..#locSize] uint(8);
+      var str_idx = (+ scan sections_sizes) - sections_sizes;
+      forall (i, column, si) in zip(0..#ncols, sym_names, str_idx) {
+        // generate the local c string list of column names
+        c_names[i] = my_column_names[i].localize().c_str();
+
+        var entry = st.lookup(column);
+
+        // access the dtype of each 
+        var entryDtype = DType.UNDEF;
+        if (entry.isAssignableTo(SymbolEntryType.TypedArraySymEntry)) {
+          entryDtype = (entry: borrowed GenSymEntry).dtype;
+        } else if (entry.isAssignableTo(SymbolEntryType.SegStringSymEntry)) {
+          entryDtype = (entry: borrowed SegStringSymEntry).dtype;
+        } else {
+          throw getErrorWithContext(
+              msg="Unknown SymEntry Type",
+              lineNumber=getLineNumber(), 
+              routineName=getRoutineName(), 
+              moduleName=getModuleName(), 
+              errorClass='ValueError'
+          );
+        }
+        
+        select entryDtype {
+          when DType.Int64 {
+            var e = toSymEntry(toGenSymEntry(entry), int);
+            var locDom = e.a.localSubdomain();
+            // set the pointer to the entry array in the list of Pointers
+            ptrList[i] = c_ptrTo(e.a[locDom]): c_void_ptr;
+            datatypes[i] = ARROWINT64;
+            sizeList[i] = locDom.size;
+          }
+          when DType.UInt64 {
+            var e = toSymEntry(toGenSymEntry(entry), uint);
+            var locDom = e.a.localSubdomain();
+            // set the pointer to the entry array in the list of Pointers
+            ptrList[i] = c_ptrTo(e.a[locDom]): c_void_ptr;
+            datatypes[i] = ARROWUINT64;
+            sizeList[i] = locDom.size;
+          }
+          when DType.Bool {
+            var e = toSymEntry(toGenSymEntry(entry), bool);
+            var locDom = e.a.localSubdomain();
+            // set the pointer to the entry array in the list of Pointers
+            ptrList[i] = c_ptrTo(e.a[locDom]): c_void_ptr;
+            datatypes[i] = ARROWBOOLEAN;
+            sizeList[i] = locDom.size;
+          } when DType.Float64 {
+            var e = toSymEntry(toGenSymEntry(entry), real);
+            var locDom = e.a.localSubdomain();
+            // set the pointer to the entry array in the list of Pointers
+            ptrList[i] = c_ptrTo(e.a[locDom]): c_void_ptr;
+            datatypes[i] = ARROWDOUBLE;
+            sizeList[i] = locDom.size;
+          } when DType.Strings {
+            var e: SegStringSymEntry = toSegStringSymEntry(entry);
+            var segStr = new SegString("", e);
+            ref ss = segStr;
+            var A = ss.offsets.a;
+            const lastOffset = A[A.domain.high];
+            const lastValIdx = ss.values.a.domain.high;
+            const locDom = ss.offsets.a.localSubdomain();
+
+            var localOffsets = A[locDom];
+            var startValIdx = localOffsets[locDom.low];
+            var endValIdx = if (lastOffset == localOffsets[locDom.high]) then lastValIdx else A[locDom.high + 1] - 1;
+            var valIdxRange = startValIdx..endValIdx;
+            ref olda = ss.values.a;
+            str_vals[si..#valIdxRange.size] = olda[valIdxRange];
+            ptrList[i] = c_ptrTo(str_vals[si]): c_void_ptr;
+            datatypes[i] = ARROWSTRING;
+            sizeList[i] = locDom.size;
+          } otherwise {
+            throw getErrorWithContext(
+                              msg="Writing Parquet files (multi-column) does not support columns of type %s".format(entryDtype),
+                              lineNumber=getLineNumber(), 
+                              routineName=getRoutineName(), 
+                              moduleName=getModuleName(), 
+                              errorClass='DataTypeError'
+            );
+          }
+        }
+      }
+      
+      // validate all elements same size
+      var numelems: int = sizeList[0];
+      if !(&& reduce (sizeList==numelems)) {
+        throw getErrorWithContext(
+              msg="Parquet columns must be the same size",
+              lineNumber=getLineNumber(), 
+              routineName=getRoutineName(), 
+              moduleName=getModuleName(), 
+              errorClass='WriteModeError'
+        );
+      }
+      var result: int = c_writeMultiColToParquet(fname.localize().c_str(), c_ptrTo(c_names), c_ptrTo(ptrList), c_ptrTo(datatypes), ncols, numelems, ROWGROUPS, compressed, c_ptrTo(pqErr.errMsg));
+    }
+    return filesExist;
+  }
+
+  proc toParquetMultiColMsg(cmd: string, msgArgs: borrowed MessageArgs, st: borrowed SymTab): MsgTuple throws {
+    const filename: string = msgArgs.getValueOf("filename");
+    const ncols: int = msgArgs.get("num_cols").getIntValue();
+
+    // get list of the names for the columns
+    var col_names: [0..#ncols] string = msgArgs.get("col_names").getList(ncols);
+
+    // get list of sym entry names holding column data
+    var sym_names: [0..#ncols] string = msgArgs.get("columns").getList(ncols);
+
+    var compressed = msgArgs.get("compressed").getBoolValue();
+
+    // Assuming all columns have same distribution, access the first to get target locales
+    var entry = st.lookup(sym_names[0]);
+
+    // access the dtype to create symentry from abstract
+    var entryDtype = DType.UNDEF;
+    if (entry.isAssignableTo(SymbolEntryType.TypedArraySymEntry)) {
+      entryDtype = (entry: borrowed GenSymEntry).dtype;
+    } else if (entry.isAssignableTo(SymbolEntryType.SegStringSymEntry)) {
+      entryDtype = (entry: borrowed SegStringSymEntry).dtype;
+    } else {
+      throw getErrorWithContext(
+          msg="Unknown SymEntry Type",
+          lineNumber=getLineNumber(), 
+          routineName=getRoutineName(), 
+          moduleName=getModuleName(), 
+          errorClass='ValueError'
+      );
+    }
+
+    var targetLocales;
+    select entryDtype {
+      when DType.Int64 {
+        var e = toSymEntry(toGenSymEntry(entry), int);
+        targetLocales = e.a.targetLocales();
+      }
+      when DType.UInt64 {
+        var e = toSymEntry(toGenSymEntry(entry), uint);
+        targetLocales = e.a.targetLocales();
+      }
+      when DType.Bool {
+        var e = toSymEntry(toGenSymEntry(entry), bool);
+        targetLocales = e.a.targetLocales();
+      } when DType.Float64 {
+        var e = toSymEntry(toGenSymEntry(entry), real);
+        targetLocales = e.a.targetLocales();
+      } when DType.Strings {
+        var e: SegStringSymEntry = toSegStringSymEntry(entry);
+        var segStr = new SegString("", e);
+        targetLocales = segStr.offsets.a.targetLocales();
+      } otherwise {
+        throw getErrorWithContext(
+                          msg="Writing Parquet files (multi-column) does not support columns of type %s".format(entryDtype),
+                          lineNumber=getLineNumber(), 
+                          routineName=getRoutineName(), 
+                          moduleName=getModuleName(), 
+                          errorClass='DataTypeError'
+        );
+      }
+    }
+
+    var warnFlag: bool;
+    try {
+      warnFlag = writeMultiColParquet(filename, col_names, ncols, sym_names, targetLocales, compressed, st);
+    } catch e: FileNotFoundError {
+      var errorMsg = "Unable to open %s for writing: %s".format(filename,e.message());
+      pqLogger.error(getModuleName(),getRoutineName(),getLineNumber(),errorMsg);
+      return new MsgTuple(errorMsg, MsgType.ERROR);
+    } catch e: WriteModeError {
+      var errorMsg = "Write mode error %s".format(e.message());
+      pqLogger.error(getModuleName(),getRoutineName(),getLineNumber(),errorMsg);
+      return new MsgTuple(errorMsg, MsgType.ERROR);
+    } catch e: Error {
+      var errorMsg = "problem writing to file %s".format(e.message());
+      pqLogger.error(getModuleName(),getRoutineName(),getLineNumber(),errorMsg);
+      return new MsgTuple(errorMsg, MsgType.ERROR);
+    }
+
+    if warnFlag {
+      var warnMsg = "Warning: possibly overwriting existing files matching filename pattern";
+      return new MsgTuple(warnMsg, MsgType.WARNING);
+    } else {
+      var repMsg = "wrote array to file";
+      pqLogger.debug(getModuleName(),getRoutineName(),getLineNumber(),repMsg);
+      return new MsgTuple(repMsg, MsgType.NORMAL);
+    }
+  }
+
   proc lspqMsg(cmd: string, msgArgs: borrowed MessageArgs, st: borrowed SymTab): MsgTuple throws {
     // reqMsg: "lshdf [<json_filename>]"
     var repMsg: string;
@@ -938,6 +1178,7 @@ module ParquetMsg {
 
   use CommandMap;
   registerFunction("readAllParquet", readAllParquetMsg, getModuleName());
+  registerFunction("toParquet_multi", toParquetMultiColMsg, getModuleName());
   registerFunction("writeParquet", toparquetMsg, getModuleName());
   registerFunction("lspq", lspqMsg, getModuleName());
   registerFunction("getnullparquet", nullIndicesMsg, getModuleName());


Test patch:
diff --git a/tests/dataframe_test.py b/tests/dataframe_test.py
index 06c9f83e8c..23ad865b98 100644
--- a/tests/dataframe_test.py
+++ b/tests/dataframe_test.py
@@ -528,7 +528,7 @@ def test_save(self):
             akdf.to_parquet(f"{tmp_dirname}/testName")
 
             ak_loaded = ak.DataFrame.load(f"{tmp_dirname}/testName")
-            self.assertTrue(validation_df.equals(ak_loaded.to_pandas()))
+            self.assertTrue(validation_df.equals(ak_loaded[akdf.columns].to_pandas()))
 
             # test save with index true
             akdf.to_parquet(f"{tmp_dirname}/testName_with_index.pq", index=True)
diff --git a/tests/import_export_test.py b/tests/import_export_test.py
index a6aad7039b..db8de2d28c 100644
--- a/tests/import_export_test.py
+++ b/tests/import_export_test.py
@@ -100,11 +100,10 @@ def test_export_parquet(self):
         akdf = self.build_arkouda_dataframe()
         with tempfile.TemporaryDirectory(dir=ImportExportTest.ie_test_base_tmp) as tmp_dirname:
             akdf.to_parquet(f"{tmp_dirname}/ak_write")
-            print(akdf.__repr__())
 
             pddf = ak.export(f"{tmp_dirname}/ak_write", write_file=f"{tmp_dirname}/pd_from_ak.parquet", index=True)
             self.assertEqual(len(glob.glob(f"{tmp_dirname}/pd_from_ak.parquet")), 1)
-            self.assertTrue(pddf.equals(akdf.to_pandas()))
+            self.assertTrue(pddf[akdf.columns].equals(akdf.to_pandas()))
 
             with self.assertRaises(RuntimeError):
                 pddf = ak.export(f"{tmp_dirname}/foo.h5", write_file=f"{tmp_dirname}/pd_from_ak.h5", index=True)