| 58 | double ToDouble(const Decimal256& value) const { return value.ToDouble(decimal_scale); } |
| 59 | |
| 60 | Status Consume(KernelContext*, const ExecSpan& batch) override { |
| 61 | if (!this->all_valid) return Status::OK(); |
| 62 | if (!options.skip_nulls && batch[0].null_count() > 0) { |
| 63 | this->all_valid = false; |
| 64 | return Status::OK(); |
| 65 | } |
| 66 | if (batch[0].is_array()) { |
| 67 | const ArraySpan& data = batch[0].array; |
| 68 | const CType* values = data.GetValues<CType>(1); |
| 69 | |
| 70 | if (data.length > data.GetNullCount()) { |
| 71 | this->count += data.length - data.GetNullCount(); |
| 72 | VisitSetBitRunsVoid(data.buffers[0].data, data.offset, data.length, |
| 73 | [&](int64_t pos, int64_t len) { |
| 74 | for (int64_t i = 0; i < len; ++i) { |
| 75 | this->tdigest.NanAdd(ToDouble(values[pos + i])); |
| 76 | } |
| 77 | }); |
| 78 | } |
| 79 | } else { |
| 80 | const CType value = UnboxScalar<ArrowType>::Unbox(*batch[0].scalar); |
| 81 | if (batch[0].scalar->is_valid) { |
| 82 | this->count += 1; |
| 83 | for (int64_t i = 0; i < batch.length; i++) { |
| 84 | this->tdigest.NanAdd(ToDouble(value)); |
| 85 | } |
| 86 | } |
| 87 | } |
| 88 | return Status::OK(); |
| 89 | } |
| 90 | |
| 91 | Status MergeFrom(KernelContext*, KernelState&& src) override { |
| 92 | const auto& other = checked_cast<const ThisType&>(src); |
nothing calls this directly
no test coverage detected