Status FixedWidthKeyEncoder::Encode(const ExecValue& data, int64_t batch_length,
                                    uint8_t** encoded_bytes) {
  if (data.is_array()) {
    // ...
  } else {
    const auto& scalar = data.scalar_as<arrow::internal::PrimitiveScalarBase>();
    if (scalar.is_valid) { // <-- is valid is checked here..
      const std::string_view data = scalar.view();
      DCHECK_EQ(data.size(), static_cast<size_t>(byte_width_));
      for (int64_t i = 0; i < batch_length; i++) {
        auto& encoded_ptr = *encoded_bytes++;
        *encoded_ptr++ = kValidByte;
        memcpy(encoded_ptr, data.data(), data.size());
        encoded_ptr += byte_width_;
      }
    } else {
      for (int64_t i = 0; i < batch_length; i++) {
        auto& encoded_ptr = *encoded_bytes++;
        *encoded_ptr++ = kNullByte;
        memset(encoded_ptr, 0, byte_width_);
        encoded_ptr += byte_width_;
      }
    }
  }
  return Status::OK();
}

Status BooleanKeyEncoder::Encode(const ExecValue& data, int64_t batch_length,
                                 uint8_t** encoded_bytes) {
  if (data.is_array()) {
    // ..
  } else {
    const auto& scalar = data.scalar_as<BooleanScalar>();
    bool value = scalar.is_valid && scalar.value;  <-- boolean is handled in this way
    for (int64_t i = 0; i < batch_length; i++) {
      auto& encoded_ptr = *encoded_bytes++;
      *encoded_ptr++ = kValidByte;
      *encoded_ptr++ = value;
    }
  }
  return Status::OK();
}