Load schema files from hdfs*/
| 239 | |
| 240 | /* Load schema files from hdfs*/ |
| 241 | void reloadFormatSchema(ContextMutablePtr context, String remote_format_schema_path, String format_schema_path, Poco::Logger * log) |
| 242 | { |
| 243 | #if USE_HDFS |
| 244 | if (!remote_format_schema_path.empty()) |
| 245 | { |
| 246 | remote_format_schema_path += "/"; // add it by default |
| 247 | // try download files from remote_format_schema_path to format_schema_path |
| 248 | Poco::URI remote_uri(remote_format_schema_path); |
| 249 | if (isHdfsOrCfsScheme(remote_uri.getScheme())) |
| 250 | { |
| 251 | HDFSBuilderPtr builder = context->getHdfsConnectionParams().createBuilder(remote_uri); |
| 252 | HDFSFSPtr fs = createHDFSFS(builder.get()); |
| 253 | int num = 0; |
| 254 | hdfsFileInfo* files = hdfsListDirectory(fs.get(), remote_uri.getPath().c_str(), &num); |
| 255 | for (int i = 0; i < num; i++) |
| 256 | { |
| 257 | String fileName(files[i].mName); |
| 258 | Poco::Path path(fileName); |
| 259 | String shortFileName = path.getFileName(); |
| 260 | String suffix = path.getExtension(); |
| 261 | if (files[i].mKind == kObjectKindDirectory || (suffix != "proto" && suffix != "capnp")) continue; // skip directory |
| 262 | Poco::File target_file(format_schema_path+ "/" + shortFileName); |
| 263 | // avoid download same file multiple times, checking size for now, it is not solid but should work online |
| 264 | if (target_file.exists() && (target_file.getSize() == UInt64(files[i].mSize))) |
| 265 | { |
| 266 | if(log) |
| 267 | { |
| 268 | |
| 269 | LOG_TRACE(log, "skip get same size remote_format_schema " + shortFileName); |
| 270 | } |
| 271 | continue; |
| 272 | } |
| 273 | |
| 274 | Poco::File file(format_schema_path+"/chtmp_" + shortFileName); |
| 275 | if (file.exists()) file.remove(); // remove last residual file |
| 276 | |
| 277 | ReadBufferFromByteHDFS reader(fileName, context->getHdfsConnectionParams()); |
| 278 | WriteBufferFromFile writer(file.path()); |
| 279 | copyData(reader, writer, nullptr); |
| 280 | if (target_file.exists()) target_file.remove(); |
| 281 | |
| 282 | file.renameTo(format_schema_path+ "/" + shortFileName); |
| 283 | if(log) |
| 284 | { |
| 285 | LOG_INFO(log, "get remote_format_schema " + shortFileName); |
| 286 | } |
| 287 | } |
| 288 | hdfsFreeFileInfo(files, num); |
| 289 | } |
| 290 | else |
| 291 | { |
| 292 | if(log) {LOG_ERROR(log, "remote_format_schema_path only support hdfs and cfs");} |
| 293 | } |
| 294 | } |
| 295 | #endif |
| 296 | return; |
| 297 | } |
| 298 |
no test coverage detected