feat: DH-22441: allow Dictionary encoding for Barrage and Arrow clients

R/rdeephaven/R/table_handle_wrapper.R

@@ -210,15 +210,15 @@ TableHandle <- R6Class("TableHandle",
     #' @return A dplyr tibble constructed from the data of this TableHandle.
     as_tibble = function() {
       rbsr <- self$as_record_batch_reader()
-      arrow_tbl <- maybe_decode_ree_columns(rbsr$read_table())
+      arrow_tbl <- maybe_decode_ree_columns(maybe_decode_dictionary_columns(rbsr$read_table()))
       return(as_tibble(arrow_tbl))
     },
 
     #' @description
     #' Converts the table referenced by this TableHandle to an R data frame.
     #' @return An R data frame constructed from the data of this TableHandle.
     as_data_frame = function() {
-      arrow_tbl <- maybe_decode_ree_columns(self$as_arrow_table())
+      arrow_tbl <- maybe_decode_ree_columns(maybe_decode_dictionary_columns(self$as_arrow_table()))
       return(as.data.frame(as.data.frame(arrow_tbl))) # TODO: for some reason as.data.frame on arrow table returns a tibble, not a data frame
     },
 
@@ -638,6 +638,19 @@ maybe_decode_ree_columns <- function(arrow_tbl) {
   return(arrow_tbl)
 }
 
+maybe_decode_dictionary_columns <- function(arrow_tbl) {
+  for (i in seq_len(arrow_tbl$num_columns)) {
+    col <- arrow_tbl$column(i - 1)
+    if (col$type$name == "dictionary") {
+      col_name <- arrow_tbl$schema$field(i - 1)$name
+      decoded <- arrow::call_function("dictionary_decode", col)
+      arrow_tbl <- arrow_tbl$SetColumn(i - 1,
+        arrow::field(col_name, decoded$type), decoded)
+    }
+  }
+  return(arrow_tbl)
+}
+
 #' @export
 as_record_batch_reader.TableHandle <- function(x, ...) {
   return(x$as_record_batch_reader())

R/rdeephaven/inst/tests/testthat/test_encoding.R

@@ -0,0 +1,78 @@
+library(testthat)
+library(rdeephaven)
+
+target <- get_dh_target()
+
+# Python script that creates REE- and dictionary-encoded tables on the server.
+# ree_table  : 6 rows, Sym = ["a","a","a","b","b","b"]
+# dict_table : 5 rows, Sym = ["x","y","z","x","y"]
+ENCODING_SETUP_SCRIPT <- '
+import jpy
+from deephaven import new_table
+from deephaven.column import string_col
+
+_JIntCls     = jpy.get_type("org.apache.arrow.vector.types.pojo.ArrowType$Int")
+_JUtf8Cls    = jpy.get_type("org.apache.arrow.vector.types.pojo.ArrowType$Utf8")
+_JREECls     = jpy.get_type("org.apache.arrow.vector.types.pojo.ArrowType$RunEndEncoded")
+_JDictEncCls = jpy.get_type("org.apache.arrow.vector.types.pojo.DictionaryEncoding")
+_JField      = jpy.get_type("org.apache.arrow.vector.types.pojo.Field")
+_JFieldType  = jpy.get_type("org.apache.arrow.vector.types.pojo.FieldType")
+_JSchema     = jpy.get_type("org.apache.arrow.vector.types.pojo.Schema")
+_JHashMap    = jpy.get_type("java.util.HashMap")
+_JArrayList  = jpy.get_type("java.util.ArrayList")
+_JInt32      = _JIntCls(32, True)
+_JUtf8       = _JUtf8Cls()
+_JREE        = _JREECls.INSTANCE
+
+def _make_ree_field(col_name, val_type, dh_type_str):
+    run_ends = _JField.notNullable("run_ends", _JInt32)
+    attrs = _JHashMap()
+    attrs.put("deephaven:type", dh_type_str)
+    val_children = _JArrayList()
+    val_f = _JField("values", _JFieldType(True, val_type, None, attrs), val_children)
+    children = _JArrayList()
+    children.add(run_ends)
+    children.add(val_f)
+    return _JField(col_name, _JFieldType(True, _JREE, None, None), children)
+
+_ree_src    = new_table([string_col("Sym", ["a", "a", "a", "b", "b", "b"])])
+_ree_fields = _JArrayList()
+_ree_fields.add(_make_ree_field("Sym", _JUtf8, "java.lang.String"))
+_ree_schema = _JSchema(_ree_fields)
+ree_table   = _ree_src.with_attributes({"BarrageSchema": _ree_schema})
+
+_dict_src    = new_table([string_col("Sym", ["x", "y", "z", "x", "y"])])
+_dict_enc    = _JDictEncCls(0, False, _JInt32)
+_dict_fields = _JArrayList()
+_dict_fields.add(_JField("Sym", _JFieldType(True, _JUtf8, _dict_enc, None), _JArrayList()))
+_dict_schema = _JSchema(_dict_fields)
+dict_table   = _dict_src.with_attributes({"BarrageSchema": _dict_schema})
+'
+
+test_that("run-end-encoded table is fetched and decoded correctly", {
+  client <- Client$new(target = target)
+  client$run_script(ENCODING_SETUP_SCRIPT)
+
+  th  <- client$open_table("ree_table")
+  df  <- as.data.frame(th)
+
+  expect_equal(nrow(df), 6)
+  expect_equal(df$Sym, c("a", "a", "a", "b", "b", "b"))
+
+  client$close()
+})
+
+test_that("dictionary-encoded table is fetched and decoded correctly", {
+  client <- Client$new(target = target)
+  client$run_script(ENCODING_SETUP_SCRIPT)
+
+  th  <- client$open_table("dict_table")
+  df  <- as.data.frame(th)
+
+  expect_equal(nrow(df), 5)
+  expect_equal(df$Sym, c("x", "y", "z", "x", "y"))
+
+  client$close()
+})
+
+rm(target, ENCODING_SETUP_SCRIPT)

cpp-client/deephaven/dhclient/src/arrowutil/arrow_array_converter.cc

@@ -13,9 +13,12 @@
 #include <vector>
 #include <arrow/visitor.h>
 #include <arrow/array/array_base.h>
+#include <arrow/array/array_dict.h>
 #include <arrow/array/array_primitive.h>
+#include <arrow/array/array_run_end.h>
 #include <arrow/scalar.h>
 #include <arrow/builder.h>
+#include <arrow/type.h>
 #include "deephaven/client/arrowutil/arrow_column_source.h"
 #include "deephaven/client/utility/arrow_util.h"
 #include "deephaven/dhcore/chunk/chunk.h"
@@ -262,6 +265,90 @@ struct ChunkedArrayToColumnSourceVisitor final : public arrow::TypeVisitor {
     return arrow::Status::OK();
   }
 
+  /**
+   * Expands a dictionary-encoded column (indices + dictionary values) to a plain-value column.
+   * Arrow Flight preserves DictionaryArrays in the RecordBatch; we decode them here so that the
+   * rest of the converter sees only the logical value type.
+   */
+  arrow::Status Visit(const arrow::DictionaryType &type) final {
+    auto dict_arrays = DowncastChunks<arrow::DictionaryArray>(*chunked_array_);
+    std::vector<std::shared_ptr<arrow::Array>> plain_chunks;
+    plain_chunks.reserve(dict_arrays.size());
+    for (const auto &da : dict_arrays) {
+      auto builder_result = arrow::MakeBuilder(type.value_type());
+      OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder_result.status()));
+      auto builder = std::move(builder_result).ValueUnsafe();
+      OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->Reserve(da->length())));
+      for (int64_t i = 0; i < da->length(); ++i) {
+        if (da->IsNull(i)) {
+          OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->AppendNull()));
+        } else {
+          // GetValueIndex handles all index widths (Int8/Int16/Int32/Int64).
+          auto scalar = ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(
+              da->dictionary()->GetScalar(da->GetValueIndex(i))));
+          OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->AppendScalar(*scalar)));
+        }
+      }
+      plain_chunks.push_back(ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->Finish())));
+    }
+    auto plain_ca = ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(
+        arrow::ChunkedArray::Make(std::move(plain_chunks), type.value_type())));
+    result_ = ArrowArrayConverter::ChunkedArrayToColumnSource(std::move(plain_ca));
+    return arrow::Status::OK();
+  }
+
+  /**
+   * Expands a run-end-encoded (REE) column to a plain-value column. REE compresses repeated values
+   * into (run_end, value) pairs; we expand them back to one element per logical row.
+   */
+  arrow::Status Visit(const arrow::RunEndEncodedType &type) final {
+    std::vector<std::shared_ptr<arrow::Array>> plain_chunks;
+    plain_chunks.reserve(chunked_array_->num_chunks());
+    for (const auto &arr : chunked_array_->chunks()) {
+      const auto &ree = static_cast<const arrow::RunEndEncodedArray &>(*arr);
+      const auto &run_ends_arr = *ree.run_ends();
+      auto builder_result = arrow::MakeBuilder(type.value_type());
+      OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder_result.status()));
+      auto builder = std::move(builder_result).ValueUnsafe();
+      OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->Reserve(ree.length())));
+      int64_t logical_pos = 0;
+      int64_t num_runs = run_ends_arr.length();
+      for (int64_t run = 0; run < num_runs; ++run) {
+        int64_t run_end;
+        switch (type.run_end_type()->id()) {
+          case arrow::Type::INT16:
+            run_end = static_cast<const arrow::Int16Array &>(run_ends_arr).Value(run);
+            break;
+          case arrow::Type::INT32:
+            run_end = static_cast<const arrow::Int32Array &>(run_ends_arr).Value(run);
+            break;
+          case arrow::Type::INT64:
+            run_end = static_cast<const arrow::Int64Array &>(run_ends_arr).Value(run);
+            break;
+          default: {
+            auto message = fmt::format("Unexpected REE run_ends type: {}",
+                type.run_end_type()->ToString());
+            throw std::runtime_error(DEEPHAVEN_LOCATION_STR(message));
+          }
+        }
+        int64_t count = run_end - logical_pos;
+        if (count <= 0) {
+          continue;
+        }
+        auto scalar = ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(ree.values()->GetScalar(run)));
+        for (int64_t j = 0; j < count; ++j) {
+          OkOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->AppendScalar(*scalar)));
+        }
+        logical_pos = run_end;
+      }
+      plain_chunks.push_back(ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(builder->Finish())));
+    }
+    auto plain_ca = ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(
+        arrow::ChunkedArray::Make(std::move(plain_chunks), type.value_type())));
+    result_ = ArrowArrayConverter::ChunkedArrayToColumnSource(std::move(plain_ca));
+    return arrow::Status::OK();
+  }
+
   /**
    * When the element is a list, we transform it into a ColumnSource<shared_ptr<ContainerBase>>.
    */

cpp-client/deephaven/dhclient/src/client.cc

@@ -32,6 +32,7 @@ using deephaven::client::server::Server;
 using deephaven::client::subscription::SubscriptionHandle;
 using deephaven::client::utility::Executor;
 using deephaven::client::utility::OkOrThrow;
+using deephaven::client::utility::ArrowUtil;
 using deephaven::client::utility::ValueOrThrow;
 using deephaven::dhcore::clienttable::ClientTable;
 using deephaven::dhcore::clienttable::Schema;
@@ -540,14 +541,14 @@ std::shared_ptr<arrow::flight::FlightStreamReader> TableHandle::GetFlightStreamR
   return GetManager().CreateFlightWrapper().GetFlightStreamReader(*this);
 }
 
-std::shared_ptr<arrow::Table> TableHandle::ToArrowTable() const {
+std::shared_ptr<ClientTable> TableHandle::ToClientTable() const {
   auto res = GetFlightStreamReader()->ToTable();
-  return ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(std::move(res)));
+  auto raw_at = ValueOrThrow(DEEPHAVEN_LOCATION_EXPR(std::move(res)));
+  return ArrowClientTable::Create(std::move(raw_at));
 }
 
-std::shared_ptr<ClientTable> TableHandle::ToClientTable() const {
-  auto at = ToArrowTable();
-  return ArrowClientTable::Create(std::move(at));
+std::shared_ptr<arrow::Table> TableHandle::ToArrowTable() const {
+  return ArrowUtil::MakeArrowTable(*ToClientTable());
 }
 
 std::shared_ptr<SubscriptionHandle> TableHandle::Subscribe(

cpp-client/deephaven/dhclient/src/utility/arrow_util.cc

@@ -141,6 +141,11 @@ struct ArrowToElementTypeId final : public arrow::TypeVisitor {
     return type.value_type()->Accept(this);
   }
 
+  // The logical element type of a dictionary-encoded column is the value type, not the index type.
+  arrow::Status Visit(const arrow::DictionaryType &type) final {
+    return type.value_type()->Accept(this);
+  }
+
   arrow::Status Visit(const arrow::Time64Type &/*type*/) final {
     element_type_ = ElementType::Of(ElementTypeId::kLocalTime);
     return arrow::Status::OK();

cpp-client/deephaven/tests/CMakeLists.txt

@@ -11,6 +11,7 @@ add_executable(dhclient_tests
         src/buffer_column_source_test.cc
         src/cython_support_test.cc
         src/date_time_test.cc
+        src/encoding_test.cc
         src/filter_test.cc
         src/group_test.cc
         src/head_and_tail_test.cc

cpp-client/deephaven/tests/src/encoding_test.cc

@@ -0,0 +1,83 @@
+/*
+ * Copyright (c) 2016-2026 Deephaven Data Labs and Patent Pending
+ */
+#include <string>
+#include <vector>
+#include "deephaven/third_party/catch.hpp"
+#include "deephaven/tests/test_util.h"
+
+using deephaven::client::utility::TableMaker;
+
+namespace deephaven::client::tests {
+
+// Python script that creates REE- and dictionary-encoded tables on the server.
+// ree_table  : 6 rows, Sym = ["a","a","a","b","b","b"]
+// dict_table : 5 rows, Sym = ["x","y","z","x","y"]
+static const char *kEncodingSetupScript = R"xxx(
+import jpy
+from deephaven import new_table
+from deephaven.column import string_col
+
+_JIntCls     = jpy.get_type('org.apache.arrow.vector.types.pojo.ArrowType$Int')
+_JUtf8Cls    = jpy.get_type('org.apache.arrow.vector.types.pojo.ArrowType$Utf8')
+_JREECls     = jpy.get_type('org.apache.arrow.vector.types.pojo.ArrowType$RunEndEncoded')
+_JDictEncCls = jpy.get_type('org.apache.arrow.vector.types.pojo.DictionaryEncoding')
+_JField      = jpy.get_type('org.apache.arrow.vector.types.pojo.Field')
+_JFieldType  = jpy.get_type('org.apache.arrow.vector.types.pojo.FieldType')
+_JSchema     = jpy.get_type('org.apache.arrow.vector.types.pojo.Schema')
+_JHashMap    = jpy.get_type('java.util.HashMap')
+_JArrayList  = jpy.get_type('java.util.ArrayList')
+_JInt32      = _JIntCls(32, True)
+_JUtf8       = _JUtf8Cls()
+_JREE        = _JREECls.INSTANCE
+
+def _make_ree_field(col_name, val_type, dh_type_str):
+    run_ends = _JField.notNullable('run_ends', _JInt32)
+    attrs = _JHashMap()
+    attrs.put('deephaven:type', dh_type_str)
+    val_children = _JArrayList()
+    val_f = _JField('values', _JFieldType(True, val_type, None, attrs), val_children)
+    children = _JArrayList()
+    children.add(run_ends)
+    children.add(val_f)
+    return _JField(col_name, _JFieldType(True, _JREE, None, None), children)
+
+_ree_src    = new_table([string_col('Sym', ['a', 'a', 'a', 'b', 'b', 'b'])])
+_ree_fields = _JArrayList()
+_ree_fields.add(_make_ree_field('Sym', _JUtf8, 'java.lang.String'))
+_ree_schema = _JSchema(_ree_fields)
+ree_table   = _ree_src.with_attributes({'BarrageSchema': _ree_schema})
+
+_dict_src    = new_table([string_col('Sym', ['x', 'y', 'z', 'x', 'y'])])
+_dict_enc    = _JDictEncCls(0, False, _JInt32)
+_dict_fields = _JArrayList()
+_dict_fields.add(_JField('Sym', _JFieldType(True, _JUtf8, _dict_enc, None), _JArrayList()))
+_dict_schema = _JSchema(_dict_fields)
+dict_table   = _dict_src.with_attributes({'BarrageSchema': _dict_schema})
+)xxx";
+
+TEST_CASE("Run-end-encoded table is fetched and decoded correctly", "[encoding]") {
+  auto client = TableMakerForTests::CreateClient();
+  auto thm = client.GetManager();
+
+  thm.RunScript(kEncodingSetupScript);
+  auto t = thm.FetchTable("ree_table");
+
+  TableMaker expected;
+  expected.AddColumn<std::string>("Sym", {"a", "a", "a", "b", "b", "b"});
+  TableComparerForTests::Compare(expected, t);
+}
+
+TEST_CASE("Dictionary-encoded table is fetched and decoded correctly", "[encoding]") {
+  auto client = TableMakerForTests::CreateClient();
+  auto thm = client.GetManager();
+
+  thm.RunScript(kEncodingSetupScript);
+  auto t = thm.FetchTable("dict_table");
+
+  TableMaker expected;
+  expected.AddColumn<std::string>("Sym", {"x", "y", "z", "x", "y"});
+  TableComparerForTests::Compare(expected, t);
+}
+
+}  // namespace deephaven::client::tests