K-way merge of per-batch sparse CSR metadata chunks (in batch_index order) into a single flat CSRMetadata with a dense global indptr. Per-batch chunks carry (srcRows, counts) sparse runs — NOT a dense global indptr — so a batch only pays for the distinct source rows it touched (the old dense representation cost numSourceRows+1 entries per batch, i.e. ~B x 800MB for a 100M-node table, which was the
| 124 | // result-construction time, so result construction stays zero-work for |
| 125 | // the NO_ORDER / INSERTION_ORDER path. |
| 126 | static ArrowQueryResult::CSRMetadata kwayMergeCSRChunks( |
| 127 | std::vector<ArrowQueryResult::CSRMetadata> chunks) { |
| 128 | if (chunks.empty()) { |
| 129 | return ArrowQueryResult::CSRMetadata{}; |
| 130 | } |
| 131 | const auto& first = chunks.front(); |
| 132 | const auto hasEdgeIDs = first.hasEdgeIDs; |
| 133 | |
| 134 | int64_t numSourceRows = 0; |
| 135 | size_t totalIndices = 0; |
| 136 | for (const auto& c : chunks) { |
| 137 | if (c.hasEdgeIDs != hasEdgeIDs) { |
| 138 | return ArrowQueryResult::CSRMetadata{}; |
| 139 | } |
| 140 | if (c.hasEdgeIDs && c.edgeIDs.size() != c.indices.size()) { |
| 141 | return ArrowQueryResult::CSRMetadata{}; |
| 142 | } |
| 143 | if (c.srcRows.size() != c.counts.size()) { |
| 144 | return ArrowQueryResult::CSRMetadata{}; |
| 145 | } |
| 146 | int64_t sumCounts = 0; |
| 147 | for (auto cnt : c.counts) { |
| 148 | if (cnt < 0) { |
| 149 | return ArrowQueryResult::CSRMetadata{}; |
| 150 | } |
| 151 | sumCounts += cnt; |
| 152 | } |
| 153 | // Sparse-run form: sum(counts) must equal indices.size(). Legacy |
| 154 | // dense form (empty srcRows, pre-set indptr): skip this check, |
| 155 | // there are no per-source counts to sum. |
| 156 | if (!c.srcRows.empty() && static_cast<size_t>(sumCounts) != c.indices.size()) { |
| 157 | return ArrowQueryResult::CSRMetadata{}; |
| 158 | } |
| 159 | if (c.numSourceRows > numSourceRows) { |
| 160 | numSourceRows = c.numSourceRows; |
| 161 | } |
| 162 | totalIndices += c.indices.size(); |
| 163 | } |
| 164 | |
| 165 | ArrowQueryResult::CSRMetadata merged; |
| 166 | merged.hasEdgeIDs = hasEdgeIDs; |
| 167 | merged.numSourceRows = numSourceRows; |
| 168 | |
| 169 | if (chunks.size() == 1) { |
| 170 | auto& c = chunks[0]; |
| 171 | merged.indices = std::move(c.indices); |
| 172 | merged.edgeIDs = std::move(c.edgeIDs); |
| 173 | // Sparse-run form (production path): rebuild the dense indptr from |
| 174 | // (srcRows, counts). Legacy dense form (test fixtures / older |
| 175 | // callers that set indptr directly with empty srcRows): keep the |
| 176 | // provided indptr as-is. |
| 177 | if (!c.srcRows.empty()) { |
| 178 | merged.indptr = buildDenseIndptr(numSourceRows, c.srcRows, c.counts); |
| 179 | if (merged.indptr.empty()) { |
| 180 | return ArrowQueryResult::CSRMetadata{}; |
| 181 | } |
| 182 | } else { |
| 183 | merged.indptr = std::move(c.indptr); |
no test coverage detected