| 169 | } // namespace |
| 170 | |
| 171 | PaimonDataSource::PaimonDataSource( |
| 172 | const std::shared_ptr<const RowType>& outputType, |
| 173 | const std::shared_ptr<ConnectorTableHandle>& tableHandle, |
| 174 | const std::unordered_map<std::string, std::shared_ptr<ColumnHandle>>& |
| 175 | columnHandles, |
| 176 | const std::shared_ptr<ConnectorQueryCtx>& queryCtx, |
| 177 | const core::QueryConfig& queryConfig, |
| 178 | const std::shared_ptr<PaimonConfig>& paimonConfig) |
| 179 | : outputType_(outputType), |
| 180 | tableHandle_(std::dynamic_pointer_cast<PaimonTableHandle>(tableHandle)), |
| 181 | expressionEvaluator_(queryCtx->expressionEvaluator()), |
| 182 | pool_(queryCtx->memoryPool()) { |
| 183 | // Wrap the query-context pool for Paimon's native memory management. |
| 184 | paimonPool_ = |
| 185 | std::make_shared<BoltPaimonMemoryPool>(pool_, expressionEvaluator_); |
| 186 | |
| 187 | ::paimon::ReadContextBuilder ctxBuilder(tableHandle_->tablePath()); |
| 188 | std::vector<std::string> columns; |
| 189 | columns.reserve(outputType_->size()); |
| 190 | std::vector<TypePtr> filterTypes; |
| 191 | filterTypes.reserve(outputType_->size()); |
| 192 | std::unordered_set<std::string> readColumnNames; |
| 193 | for (const auto& outName : outputType_->names()) { |
| 194 | auto it = columnHandles.find(outName); |
| 195 | BOLT_CHECK( |
| 196 | it != columnHandles.end(), |
| 197 | "Could not find column handle with name: {}", |
| 198 | outName); |
| 199 | const auto& columnName = paimonColumnHandle(it->second, outName)->name(); |
| 200 | columns.push_back(columnName); |
| 201 | readColumnNames.insert(columnName); |
| 202 | } |
| 203 | for (const auto& type : outputType_->children()) { |
| 204 | filterTypes.push_back(type); |
| 205 | } |
| 206 | |
| 207 | if (tableHandle_->filter()) { |
| 208 | for (const auto& fieldName : collectFieldNames(tableHandle_->filter())) { |
| 209 | auto columnHandle = findPaimonColumnHandle(columnHandles, fieldName); |
| 210 | BOLT_CHECK_NOT_NULL( |
| 211 | columnHandle, |
| 212 | "Could not find column handle for filter field: {}", |
| 213 | fieldName); |
| 214 | if (readColumnNames.insert(columnHandle->name()).second) { |
| 215 | columns.push_back(columnHandle->name()); |
| 216 | filterTypes.push_back(columnHandle->type()); |
| 217 | } |
| 218 | } |
| 219 | } |
| 220 | auto filterNames = columns; |
| 221 | filterRowType_ = ROW(std::move(filterNames), std::move(filterTypes)); |
| 222 | VLOG(1) << "PaimonDataSource::PaimonDataSource(): Read schema: " |
| 223 | << folly::join(", ", columns); |
| 224 | ctxBuilder.SetReadSchema(columns); |
| 225 | ctxBuilder.EnableMultiThreadRowToBatch(paimonConfig->multiThreadRowToBatch()); |
| 226 | if (paimonConfig->multiThreadRowToBatch()) { |
| 227 | ctxBuilder.SetRowToBatchThreadNumber( |
| 228 | paimonConfig->rowToBatchThreadNumber()); |
nothing calls this directly
no test coverage detected