diff --git a/.github/workflows/linux-build-base.yml b/.github/workflows/linux-build-base.yml index a0b86504a3e..49529a21473 100644 --- a/.github/workflows/linux-build-base.yml +++ b/.github/workflows/linux-build-base.yml @@ -52,6 +52,7 @@ jobs: - name: Build Artifact env: VELOX_DEPENDENCY_SOURCE: BUNDLED + Boost_SOURCE: SYSTEM ICU_SOURCE: SYSTEM MAKEFLAGS: MAX_HIGH_MEM_JOBS=4 MAX_LINK_JOBS=3 run: | diff --git a/CMake/resolve_dependency_modules/antlr4-runtime.cmake b/CMake/resolve_dependency_modules/antlr4-runtime.cmake new file mode 100644 index 00000000000..86bc799ce71 --- /dev/null +++ b/CMake/resolve_dependency_modules/antlr4-runtime.cmake @@ -0,0 +1,36 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +set(VELOX_ANTLR4_RUNTIME_VERSION 4.13.2) +set(VELOX_ANTLR4_RUNTIME_BUILD_SHA256_CHECKSUM + 9f18272a9b32b622835a3365f850dd1063d60f5045fb1e12ce475ae6e18a35bb) +set(VELOX_ANTLR4_RUNTIME_SOURCE_URL + "https://github.com/antlr/antlr4/archive/refs/tags/${VELOX_ANTLR4_RUNTIME_VERSION}.tar.gz" +) + +velox_resolve_dependency_url(ANTLR4_RUNTIME) + +message(STATUS "Building antlr4-runtime from source") + +FetchContent_Declare( + antlr4-runtime + URL ${VELOX_ANTLR4_RUNTIME_SOURCE_URL} + URL_HASH ${VELOX_ANTLR4_RUNTIME_BUILD_SHA256_CHECKSUM} + SOURCE_SUBDIR runtime/Cpp OVERRIDE_FIND_PACKAGE) + +set(ANTLR4_INSTALL + ON + CACHE BOOL "" FORCE) +FetchContent_MakeAvailable(antlr4-runtime) diff --git a/CMake/resolve_dependency_modules/clp.cmake b/CMake/resolve_dependency_modules/clp.cmake new file mode 100644 index 00000000000..6859f41ae08 --- /dev/null +++ b/CMake/resolve_dependency_modules/clp.cmake @@ -0,0 +1,44 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +FetchContent_Declare( + clp + GIT_REPOSITORY https://github.com/y-scope/clp.git + GIT_TAG 0798100389bd5231b520ec48ab186275795e3790) + +set(CLP_BUILD_CLP_REGEX_UTILS + OFF + CACHE BOOL "Build CLP regex utils") +set(CLP_BUILD_CLP_S_JSONCONSTRUCTOR + OFF + CACHE BOOL "Build CLP-S JSON constructor") +set(CLP_BUILD_CLP_S_REDUCER_DEPENDENCIES + OFF + CACHE BOOL "Build CLP-S reducer dependencies") +set(CLP_BUILD_CLP_S_SEARCH_SQL + OFF + CACHE BOOL "Build CLP-S search SQL") +set(CLP_BUILD_EXECUTABLES + OFF + CACHE BOOL "Build CLP executables") +set(CLP_BUILD_TESTING + OFF + CACHE BOOL "Build CLP tests") + +FetchContent_Populate(clp) + +list(APPEND CMAKE_MODULE_PATH "${clp_SOURCE_DIR}/components/core/cmake/Modules") +add_subdirectory(${clp_SOURCE_DIR}/components/core + ${clp_BINARY_DIR}/components/core) diff --git a/CMake/resolve_dependency_modules/date.cmake b/CMake/resolve_dependency_modules/date.cmake new file mode 100644 index 00000000000..275aa6e0a08 --- /dev/null +++ b/CMake/resolve_dependency_modules/date.cmake @@ -0,0 +1,38 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +set(VELOX_DATE_BUILD_VERSION 3.0.1) +set(VELOX_DATE_BUILD_SHA256_CHECKSUM + 7a390f200f0ccd207e8cff6757e04817c1a0aec3e327b006b7eb451c57ee3538) +set(VELOX_DATE_SOURCE_URL + "https://github.com/HowardHinnant/date/archive/refs/tags/v${VELOX_DATE_BUILD_VERSION}.tar.gz" +) + +velox_resolve_dependency_url(DATE) + +# Optionally set CMake variables *before* make-available +set(CMAKE_INSTALL_MESSAGE + LAZY + CACHE STRING "" FORCE) + +message(STATUS "Building date from source") + +FetchContent_Declare( + date + URL ${VELOX_DATE_SOURCE_URL} + URL_HASH ${VELOX_DATE_BUILD_SHA256_CHECKSUM} + OVERRIDE_FIND_PACKAGE) + +FetchContent_MakeAvailable(date) diff --git a/CMake/resolve_dependency_modules/msgpack-cxx.cmake b/CMake/resolve_dependency_modules/msgpack-cxx.cmake new file mode 100644 index 00000000000..9f20800b74e --- /dev/null +++ b/CMake/resolve_dependency_modules/msgpack-cxx.cmake @@ -0,0 +1,35 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +set(VELOX_MSGPACK_BUILD_VERSION cpp-7.0.0) +set(VELOX_MSGPACK_BUILD_SHA256_CHECKSUM + 070881ebea9208cf7e731fd5a46a11404025b2f260ab9527e32dfcb7c689fbfc) +set(VELOX_MSGPACK_SOURCE_URL + "https://github.com/msgpack/msgpack-c/archive/refs/tags/${VELOX_MSGPACK_BUILD_VERSION}.tar.gz" +) + +velox_resolve_dependency_url(MSGPACK) + +message(STATUS "Building msgpack-cxx from source") + +FetchContent_Declare( + msgpack-cxx + URL ${VELOX_MSGPACK_SOURCE_URL} + URL_HASH ${VELOX_MSGPACK_BUILD_SHA256_CHECKSUM} + OVERRIDE_FIND_PACKAGE EXCLUDE_FROM_ALL SYSTEM) + +set(MSGPACK_USE_BOOST OFF) + +FetchContent_MakeAvailable(msgpack-cxx) diff --git a/CMake/resolve_dependency_modules/nlohmann_json.cmake b/CMake/resolve_dependency_modules/nlohmann_json.cmake new file mode 100644 index 00000000000..1b0298f6618 --- /dev/null +++ b/CMake/resolve_dependency_modules/nlohmann_json.cmake @@ -0,0 +1,37 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +set(VELOX_NLOHMANN_JSON_BUILD_VERSION 3.11.3) +set(VELOX_NLOHMANN_JSON_BUILD_SHA256_CHECKSUM + 0d8ef5af7f9794e3263480193c491549b2ba6cc74bb018906202ada498a79406) +set(VELOX_NLOHMANN_JSON_SOURCE_URL + "https://github.com/nlohmann/json/archive/refs/tags/v${VELOX_NLOHMANN_JSON_BUILD_VERSION}.tar.gz" +) + +velox_resolve_dependency_url(NLOHMANN_JSON) + +message(STATUS "Building nlohmann_json from source") + +FetchContent_Declare( + nlohmann_json + URL ${VELOX_NLOHMANN_JSON_SOURCE_URL} + URL_HASH ${VELOX_NLOHMANN_JSON_BUILD_SHA256_CHECKSUM} + OVERRIDE_FIND_PACKAGE) + +set(JSON_BuildTests + OFF + CACHE INTERNAL "") + +FetchContent_MakeAvailable(nlohmann_json) diff --git a/CMake/resolve_dependency_modules/simdjson.cmake b/CMake/resolve_dependency_modules/simdjson.cmake index a0c55c3758a..5d28789747e 100644 --- a/CMake/resolve_dependency_modules/simdjson.cmake +++ b/CMake/resolve_dependency_modules/simdjson.cmake @@ -27,7 +27,8 @@ message(STATUS "Building simdjson from source") FetchContent_Declare( simdjson URL ${VELOX_SIMDJSON_SOURCE_URL} - URL_HASH ${VELOX_SIMDJSON_BUILD_SHA256_CHECKSUM}) + URL_HASH ${VELOX_SIMDJSON_BUILD_SHA256_CHECKSUM} + OVERRIDE_FIND_PACKAGE) if(${VELOX_SIMDJSON_SKIPUTF8VALIDATION}) set(SIMDJSON_SKIPUTF8VALIDATION ON) diff --git a/CMake/resolve_dependency_modules/spdlog.cmake b/CMake/resolve_dependency_modules/spdlog.cmake new file mode 100644 index 00000000000..a7a1ddfe698 --- /dev/null +++ b/CMake/resolve_dependency_modules/spdlog.cmake @@ -0,0 +1,34 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +set(VELOX_SPDLOG_BUILD_VERSION 1.12.0) +set(VELOX_SPDLOG_BUILD_SHA256_CHECKSUM + 4dccf2d10f410c1e2feaff89966bfc49a1abb29ef6f08246335b110e001e09a9) +set(VELOX_SPDLOG_SOURCE_URL + "https://github.com/gabime/spdlog/archive/refs/tags/v${VELOX_SPDLOG_BUILD_VERSION}.tar.gz" +) + +velox_resolve_dependency_url(SPDLOG) + +message(STATUS "Building spdlog from source") + +FetchContent_Declare( + spdlog + URL ${VELOX_SPDLOG_SOURCE_URL} + URL_HASH ${VELOX_SPDLOG_BUILD_SHA256_CHECKSUM} + OVERRIDE_FIND_PACKAGE EXCLUDE_FROM_ALL SYSTEM) + +set(SPDLOG_FMT_EXTERNAL ON) +FetchContent_MakeAvailable(spdlog) diff --git a/CMake/resolve_dependency_modules/ystdlib_cpp.cmake b/CMake/resolve_dependency_modules/ystdlib_cpp.cmake new file mode 100644 index 00000000000..a47b24416c8 --- /dev/null +++ b/CMake/resolve_dependency_modules/ystdlib_cpp.cmake @@ -0,0 +1,23 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +include_guard(GLOBAL) + +FetchContent_Declare( + ystdlib_cpp + GIT_REPOSITORY https://github.com/y-scope/ystdlib-cpp.git + GIT_TAG 0ae886c6a7ee706a3c6e1950262b63d72f71fe63) + +FetchContent_Populate(ystdlib_cpp) + +set(CLP_YSTDLIB_SOURCE_DIRECTORY "${ystdlib_cpp_SOURCE_DIR}") diff --git a/CMakeLists.txt b/CMakeLists.txt index cc27d5e5623..ba0734c6ee4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -127,6 +127,7 @@ set(VELOX_GFLAGS_TYPE ) option(VELOX_ENABLE_EXEC "Build exec." ON) option(VELOX_ENABLE_AGGREGATES "Build aggregates." ON) +option(VELOX_ENABLE_CLP_CONNECTOR "Build CLP connector." ON) option(VELOX_ENABLE_HIVE_CONNECTOR "Build Hive connector." ON) option(VELOX_ENABLE_TPCH_CONNECTOR "Build TPC-H connector." ON) option(VELOX_ENABLE_PRESTO_FUNCTIONS "Build Presto SQL functions." ON) @@ -174,6 +175,7 @@ if(${VELOX_BUILD_MINIMAL} OR ${VELOX_BUILD_MINIMAL_WITH_DWIO}) set(VELOX_ENABLE_EXPRESSION ON) set(VELOX_ENABLE_EXEC OFF) set(VELOX_ENABLE_AGGREGATES OFF) + set(VELOX_ENABLE_CLP_CONNECTOR OFF) set(VELOX_ENABLE_HIVE_CONNECTOR OFF) set(VELOX_ENABLE_TPCH_CONNECTOR OFF) set(VELOX_ENABLE_SPARK_FUNCTIONS OFF) @@ -459,9 +461,11 @@ set(BOOST_INCLUDE_LIBRARIES context date_time filesystem + iostreams program_options regex system + url thread) velox_set_source(Boost) @@ -518,6 +522,35 @@ endif() velox_set_source(re2) velox_resolve_dependency(re2) +if(${VELOX_ENABLE_CLP_CONNECTOR}) + velox_set_source(absl) + velox_resolve_dependency(absl) + + velox_set_source(antlr4-runtime) + velox_resolve_dependency(antlr4-runtime) + + velox_set_source(date) + velox_resolve_dependency(date) + + velox_set_source(msgpack-cxx) + velox_resolve_dependency(msgpack-cxx) + + velox_set_source(nlohmann_json) + velox_resolve_dependency(nlohmann_json) + + velox_set_source(simdjson) + velox_resolve_dependency(simdjson) + + velox_set_source(spdlog) + velox_resolve_dependency(spdlog) + + set(ystdlib_cpp_SOURCE BUNDLED) + velox_resolve_dependency(ystdlib_cpp) + + set(clp_SOURCE BUNDLED) + velox_resolve_dependency(clp) +endif() + if(${VELOX_BUILD_PYTHON_PACKAGE}) find_package( Python 3.9 diff --git a/velox/connectors/CMakeLists.txt b/velox/connectors/CMakeLists.txt index 3cc600201f6..743ecbcaac0 100644 --- a/velox/connectors/CMakeLists.txt +++ b/velox/connectors/CMakeLists.txt @@ -17,6 +17,10 @@ velox_link_libraries(velox_connector velox_common_config velox_vector) add_subdirectory(fuzzer) +if(${VELOX_ENABLE_CLP_CONNECTOR}) + add_subdirectory(clp) +endif() + if(${VELOX_ENABLE_HIVE_CONNECTOR}) add_subdirectory(hive) endif() diff --git a/velox/connectors/clp/CMakeLists.txt b/velox/connectors/clp/CMakeLists.txt new file mode 100644 index 00000000000..5cf5366aa9e --- /dev/null +++ b/velox/connectors/clp/CMakeLists.txt @@ -0,0 +1,31 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +#set(CLP_SRC_DIR ${clp_SOURCE_DIR}/components/core/src) +add_subdirectory(search_lib) + +velox_add_library( + velox_clp_connector + OBJECT + ClpConfig.cpp + ClpConnector.cpp + ClpDataSource.cpp + ClpTableHandle.cpp) + +velox_link_libraries(velox_clp_connector + PRIVATE clp-s-search simdjson::simdjson velox_connector) +target_compile_features(velox_clp_connector PRIVATE cxx_std_20) + +if(${VELOX_BUILD_TESTING}) + add_subdirectory(tests) +endif() diff --git a/velox/connectors/clp/ClpColumnHandle.h b/velox/connectors/clp/ClpColumnHandle.h new file mode 100644 index 00000000000..a4a1b8702f5 --- /dev/null +++ b/velox/connectors/clp/ClpColumnHandle.h @@ -0,0 +1,58 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "velox/connectors/Connector.h" + +namespace facebook::velox::connector::clp { + +class ClpColumnHandle : public ColumnHandle { + public: + ClpColumnHandle( + const std::string& columnName, + const std::string& originalColumnName, + const TypePtr& columnType, + bool nullable) + : columnName_(columnName), + originalColumnName_(originalColumnName), + columnType_(columnType), + nullable_(nullable) {} + + const std::string& columnName() const { + return columnName_; + } + + const std::string& originalColumnName() const { + return originalColumnName_; + } + + const TypePtr& columnType() const { + return columnType_; + } + + bool nullable() const { + return nullable_; + } + + private: + const std::string columnName_; + const std::string originalColumnName_; + const TypePtr columnType_; + const bool nullable_; +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpConfig.cpp b/velox/connectors/clp/ClpConfig.cpp new file mode 100644 index 00000000000..adbb8f0fc47 --- /dev/null +++ b/velox/connectors/clp/ClpConfig.cpp @@ -0,0 +1,42 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include "velox/connectors/clp/ClpConfig.h" + +namespace facebook::velox::connector::clp { + +namespace { + +ClpConfig::StorageType stringToStorageType(const std::string& strValue) { + auto upperValue = boost::algorithm::to_upper_copy(strValue); + if (upperValue == "FS") { + return ClpConfig::StorageType::kFs; + } + if (upperValue == "S3") { + return ClpConfig::StorageType::kS3; + } + VELOX_UNSUPPORTED("Unsupported storage type: {}.", strValue); +} + +} // namespace + +ClpConfig::StorageType ClpConfig::storageType() const { + return stringToStorageType(config_->get(kStorageType, "FS")); +} + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpConfig.h b/velox/connectors/clp/ClpConfig.h new file mode 100644 index 00000000000..ce4808eef62 --- /dev/null +++ b/velox/connectors/clp/ClpConfig.h @@ -0,0 +1,52 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "velox/common/config/Config.h" + +namespace facebook::velox::config { +class ConfigBase; +} + +namespace facebook::velox::connector::clp { + +class ClpConfig { + public: + enum class StorageType { + kFs, + kS3, + }; + + static constexpr const char* kStorageType = "clp.storage-type"; + + explicit ClpConfig(std::shared_ptr config) { + VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization"); + config_ = std::move(config); + } + + [[nodiscard]] const std::shared_ptr& config() + const { + return config_; + } + + StorageType storageType() const; + + private: + std::shared_ptr config_; +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpConnector.cpp b/velox/connectors/clp/ClpConnector.cpp new file mode 100644 index 00000000000..25f90171c27 --- /dev/null +++ b/velox/connectors/clp/ClpConnector.cpp @@ -0,0 +1,62 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "clp_s/TimestampPattern.hpp" + +#include "velox/connectors/clp/ClpConnector.h" +#include "velox/connectors/clp/ClpDataSource.h" + +namespace facebook::velox::connector::clp { + +ClpConnector::ClpConnector( + const std::string& id, + std::shared_ptr config) + : Connector(id), config_(std::make_shared(std::move(config))) {} + +std::unique_ptr ClpConnector::createDataSource( + const RowTypePtr& outputType, + const std::shared_ptr& tableHandle, + const std::unordered_map< + std::string, + std::shared_ptr>& columnHandles, + ConnectorQueryCtx* connectorQueryCtx) { + return std::make_unique( + outputType, + tableHandle, + columnHandles, + connectorQueryCtx->memoryPool(), + config_); +} + +std::unique_ptr ClpConnector::createDataSink( + RowTypePtr inputType, + std::shared_ptr connectorInsertTableHandle, + ConnectorQueryCtx* connectorQueryCtx, + CommitStrategy commitStrategy) { + VELOX_NYI("createDataSink for ClpConnector is not implemented!"); +} + +ClpConnectorFactory::ClpConnectorFactory() + : ConnectorFactory(kClpConnectorName) { + clp_s::TimestampPattern::init(); +} + +ClpConnectorFactory::ClpConnectorFactory(const char* connectorName) + : ConnectorFactory(connectorName) { + clp_s::TimestampPattern::init(); +} + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpConnector.h b/velox/connectors/clp/ClpConnector.h new file mode 100644 index 00000000000..59efe8d8bb7 --- /dev/null +++ b/velox/connectors/clp/ClpConnector.h @@ -0,0 +1,77 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "velox/connectors/Connector.h" +#include "velox/connectors/clp/ClpConfig.h" + +namespace facebook::velox::connector::clp { + +class ClpConnector : public Connector { + public: + ClpConnector( + const std::string& id, + std::shared_ptr config); + + [[nodiscard]] const std::shared_ptr& + connectorConfig() const override { + return config_->config(); + } + + [[nodiscard]] bool canAddDynamicFilter() const override { + return false; + } + + std::unique_ptr createDataSource( + const RowTypePtr& outputType, + const std::shared_ptr& tableHandle, + const std::unordered_map< + std::string, + std::shared_ptr>& columnHandles, + ConnectorQueryCtx* connectorQueryCtx) override; + + bool supportsSplitPreload() override { + return false; + } + + std::unique_ptr createDataSink( + RowTypePtr inputType, + std::shared_ptr connectorInsertTableHandle, + ConnectorQueryCtx* connectorQueryCtx, + CommitStrategy commitStrategy) override; + + private: + std::shared_ptr config_; +}; + +class ClpConnectorFactory : public ConnectorFactory { + public: + static constexpr const char* kClpConnectorName = "clp"; + + ClpConnectorFactory(); + explicit ClpConnectorFactory(const char* connectorName); + + std::shared_ptr newConnector( + const std::string& id, + std::shared_ptr config, + folly::Executor* /*ioExecutor*/, + folly::Executor* /*cpuExecutor*/) override { + return std::make_shared(id, config); + } +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpConnectorSplit.h b/velox/connectors/clp/ClpConnectorSplit.h new file mode 100644 index 00000000000..cbe13e6318b --- /dev/null +++ b/velox/connectors/clp/ClpConnectorSplit.h @@ -0,0 +1,34 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "velox/connectors/Connector.h" + +namespace facebook::velox::connector::clp { + +struct ClpConnectorSplit : public connector::ConnectorSplit { + ClpConnectorSplit(const std::string& connectorId, const std::string& path) + : connector::ConnectorSplit(connectorId), path_(path) {} + + [[nodiscard]] std::string toString() const override { + return fmt::format("CLP Split: {}", path_); + } + + const std::string path_; +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp new file mode 100644 index 00000000000..6bc4d1d1414 --- /dev/null +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -0,0 +1,176 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include "velox/connectors/clp/ClpColumnHandle.h" +#include "velox/connectors/clp/ClpConnectorSplit.h" +#include "velox/connectors/clp/ClpDataSource.h" +#include "velox/connectors/clp/ClpTableHandle.h" +#include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/connectors/clp/search_lib/ClpVectorLoader.h" +#include "velox/vector/FlatVector.h" + +namespace facebook::velox::connector::clp { + +ClpDataSource::ClpDataSource( + const RowTypePtr& outputType, + const std::shared_ptr& tableHandle, + const std::unordered_map< + std::string, + std::shared_ptr>& columnHandles, + velox::memory::MemoryPool* pool, + std::shared_ptr& clpConfig) + : pool_(pool), outputType_(outputType) { + auto clpTableHandle = std::dynamic_pointer_cast(tableHandle); + storageType_ = clpConfig->storageType(); + if (auto query = clpTableHandle->kqlQuery(); query && !query->empty()) { + kqlQuery_ = *query; + } else { + kqlQuery_ = "*"; + } + + for (const auto& outputName : outputType->names()) { + auto columnHandle = columnHandles.find(outputName); + VELOX_CHECK( + columnHandle != columnHandles.end(), + "ColumnHandle not found for output name: {}", + outputName); + auto clpColumnHandle = + std::dynamic_pointer_cast(columnHandle->second); + VELOX_CHECK_NOT_NULL( + clpColumnHandle, + "ColumnHandle must be an instance of ClpColumnHandle for output name: {}", + outputName); + auto columnName = clpColumnHandle->originalColumnName(); + auto columnType = clpColumnHandle->columnType(); + addFieldsRecursively(columnType, columnName); + } +} + +void ClpDataSource::addFieldsRecursively( + const TypePtr& columnType, + const std::string& parentName) { + if (columnType->kind() == TypeKind::ROW) { + const auto& rowType = columnType->asRow(); + for (uint32_t i = 0; i < rowType.size(); ++i) { + const auto& childType = rowType.childAt(i); + const auto childName = parentName + "." + rowType.nameOf(i); + addFieldsRecursively(childType, childName); + } + } else { + search_lib::ColumnType clpColumnType = search_lib::ColumnType::Unknown; + switch (columnType->kind()) { + case TypeKind::BOOLEAN: + clpColumnType = search_lib::ColumnType::Boolean; + break; + case TypeKind::INTEGER: + case TypeKind::BIGINT: + case TypeKind::SMALLINT: + case TypeKind::TINYINT: + clpColumnType = search_lib::ColumnType::Integer; + break; + case TypeKind::DOUBLE: + case TypeKind::REAL: + clpColumnType = search_lib::ColumnType::Float; + break; + case TypeKind::VARCHAR: + clpColumnType = search_lib::ColumnType::String; + break; + case TypeKind::ARRAY: + clpColumnType = search_lib::ColumnType::Array; + break; + default: + VELOX_USER_FAIL("Type not supported: {}", columnType->name()); + } + fields_.emplace_back(search_lib::Field{clpColumnType, parentName}); + } +} + +void ClpDataSource::addSplit(std::shared_ptr split) { + auto clpSplit = std::dynamic_pointer_cast(split); + + if (storageType_ == ClpConfig::StorageType::kFs) { + cursor_ = std::make_unique( + clp_s::InputSource::Filesystem, clpSplit->path_); + } else if (storageType_ == ClpConfig::StorageType::kS3) { + cursor_ = std::make_unique( + clp_s::InputSource::Network, clpSplit->path_); + } + + cursor_->executeQuery(kqlQuery_, fields_); +} + +VectorPtr ClpDataSource::createVector( + const TypePtr& vectorType, + size_t vectorSize, + const std::vector& projectedColumns, + const std::shared_ptr>& filteredRows, + size_t& readerIndex) { + if (vectorType->kind() == TypeKind::ROW) { + std::vector children; + auto& rowType = vectorType->as(); + for (uint32_t i = 0; i < rowType.size(); ++i) { + children.push_back(createVector( + rowType.childAt(i), + vectorSize, + projectedColumns, + filteredRows, + readerIndex)); + } + return std::make_shared( + pool_, vectorType, nullptr, vectorSize, std::move(children)); + } + auto vector = BaseVector::create(vectorType, vectorSize, pool_); + vector->setNulls(allocateNulls(vectorSize, pool_, bits::kNull)); + + VELOX_CHECK_LT( + readerIndex, projectedColumns.size(), "Reader index out of bounds"); + auto projectedColumn = projectedColumns[readerIndex]; + auto projectedType = fields_[readerIndex].type; + readerIndex++; + return std::make_shared( + pool_, + vectorType, + vectorSize, + std::make_unique( + projectedColumn, projectedType, filteredRows), + std::move(vector)); +} + +std::optional ClpDataSource::next( + uint64_t size, + ContinueFuture& future) { + auto filteredRows = std::make_shared>(); + auto rowsScanned = cursor_->fetchNext(size, filteredRows); + auto rowsFiltered = filteredRows->size(); + if (rowsFiltered == 0) { + return nullptr; + } + completedRows_ += rowsScanned; + size_t readerIndex = 0; + const auto& projectedColumns = cursor_->getProjectedColumns(); + VELOX_CHECK_EQ( + projectedColumns.size(), + fields_.size(), + "Projected columns size {} does not match fields size {}", + projectedColumns.size(), + fields_.size()); + return std::dynamic_pointer_cast(createVector( + outputType_, rowsFiltered, projectedColumns, filteredRows, readerIndex)); +} + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpDataSource.h b/velox/connectors/clp/ClpDataSource.h new file mode 100644 index 00000000000..c9ab8ea7d0f --- /dev/null +++ b/velox/connectors/clp/ClpDataSource.h @@ -0,0 +1,107 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include + +#include "velox/connectors/Connector.h" +#include "velox/connectors/clp/ClpConfig.h" +#include "velox/connectors/clp/search_lib/ClpCursor.h" + +namespace clp_s { +class BaseColumnReader; +} // namespace clp_s + +namespace facebook::velox::connector::clp { + +class ClpDataSource : public DataSource { + public: + ClpDataSource( + const RowTypePtr& outputType, + const std::shared_ptr& tableHandle, + const std::unordered_map< + std::string, + std::shared_ptr>& columnHandles, + velox::memory::MemoryPool* pool, + std::shared_ptr& clpConfig); + + void addSplit(std::shared_ptr split) override; + + std::optional next(uint64_t size, velox::ContinueFuture& future) + override; + + void addDynamicFilter( + column_index_t outputChannel, + const std::shared_ptr& filter) override { + VELOX_NYI("Dynamic filters not supported by ClpConnector."); + } + + uint64_t getCompletedBytes() override { + return completedBytes_; + } + + uint64_t getCompletedRows() override { + return completedRows_; + } + + std::unordered_map runtimeStats() override { + return {}; + } + + private: + /// Recursively adds fields from the column type to the list of fields to be + /// retrieved from the data source. + /// + /// @param columnType + /// @param parentName The name of the parent field (used for nested fields). + void addFieldsRecursively( + const TypePtr& columnType, + const std::string& parentName); + + /// Creates a Vector of the specified type and size. + /// + /// This method recursively creates vectors for complex types like ROW. For + /// primitive types, it creates a LazyVector that will load the data from the + /// underlying data source when it is accessed. + /// + /// @param vectorType + /// @param vectorSize + /// @param projectedColumns The readers of the projected columns. + /// @param filteredRows The rows to be read. + /// @param readerIndex The index of the column reader. + /// @return A Vector of the specified type and size. + VectorPtr createVector( + const TypePtr& vectorType, + size_t vectorSize, + const std::vector& projectedColumns, + const std::shared_ptr>& filteredRows, + size_t& readerIndex); + + ClpConfig::StorageType storageType_; + std::string kqlQuery_; + velox::memory::MemoryPool* pool_; + RowTypePtr outputType_; + std::set columnUntypedNames_; + uint64_t completedRows_{0}; + uint64_t completedBytes_{0}; + + std::vector fields_; + + std::unique_ptr cursor_; +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpTableHandle.cpp b/velox/connectors/clp/ClpTableHandle.cpp new file mode 100644 index 00000000000..0ef944768d8 --- /dev/null +++ b/velox/connectors/clp/ClpTableHandle.cpp @@ -0,0 +1,29 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "velox/connectors/clp/ClpTableHandle.h" + +namespace facebook::velox::connector::clp { + +std::string ClpTableHandle::toString() const { + return ConnectorTableHandle::toString(); +} + +folly::dynamic ClpTableHandle::serialize() const { + return ConnectorTableHandle::serialize(); +} + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/ClpTableHandle.h b/velox/connectors/clp/ClpTableHandle.h new file mode 100644 index 00000000000..94c917ed1b9 --- /dev/null +++ b/velox/connectors/clp/ClpTableHandle.h @@ -0,0 +1,50 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "velox/connectors/Connector.h" + +namespace facebook::velox::connector::clp { + +class ClpTableHandle : public ConnectorTableHandle { + public: + ClpTableHandle( + const std::string& connectorId, + const std::string& tableName, + std::shared_ptr kqlQuery) + : ConnectorTableHandle(connectorId), + tableName_(tableName), + kqlQuery_(std::move(kqlQuery)) {} + + [[nodiscard]] const std::string& tableName() const { + return tableName_; + } + + [[nodiscard]] const std::shared_ptr& kqlQuery() const { + return kqlQuery_; + } + + std::string toString() const override; + + folly::dynamic serialize() const override; + + private: + const std::string tableName_; + std::shared_ptr kqlQuery_; +}; + +} // namespace facebook::velox::connector::clp diff --git a/velox/connectors/clp/search_lib/CMakeLists.txt b/velox/connectors/clp/search_lib/CMakeLists.txt new file mode 100644 index 00000000000..7f163d0aa2a --- /dev/null +++ b/velox/connectors/clp/search_lib/CMakeLists.txt @@ -0,0 +1,34 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +velox_add_library( + clp-s-search + STATIC + ClpCursor.cpp + ClpCursor.h + ClpQueryRunner.cpp + ClpQueryRunner.h + ClpVectorLoader.cpp + ClpVectorLoader.h) + +velox_link_libraries( + clp-s-search + PUBLIC clp_s::archive_reader + PRIVATE + clp_s::archive_writer + clp_s::clp_dependencies + clp_s::io + clp_s::search + clp_s::search::kql + velox_vector) +target_compile_features(clp-s-search PRIVATE cxx_std_20) diff --git a/velox/connectors/clp/search_lib/ClpCursor.cpp b/velox/connectors/clp/search_lib/ClpCursor.cpp new file mode 100644 index 00000000000..bab940c92c3 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpCursor.cpp @@ -0,0 +1,257 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include "clp_s/ArchiveReader.hpp" +#include "clp_s/search/EvaluateTimestampIndex.hpp" +#include "clp_s/search/ast/ConvertToExists.hpp" +#include "clp_s/search/ast/EmptyExpr.hpp" +#include "clp_s/search/ast/NarrowTypes.hpp" +#include "clp_s/search/ast/OrOfAndForm.hpp" +#include "clp_s/search/ast/SearchUtils.hpp" +#include "clp_s/search/kql/kql.hpp" + +#include "velox/connectors/clp/search_lib/ClpCursor.h" + +using namespace clp_s; +using namespace clp_s::search; +using namespace clp_s::search::ast; + +namespace facebook::velox::connector::clp::search_lib { + +ClpCursor::ClpCursor(InputSource inputSource, std::string archivePath) + : errorCode_(ErrorCode::QueryNotInitialized), + inputSource_(inputSource), + archivePath_(std::move(archivePath)), + archiveReader_(std::make_shared()) {} + +ClpCursor::~ClpCursor() { + if (currentArchiveLoaded_) { + archiveReader_->close(); + } +} + +void ClpCursor::executeQuery( + const std::string& query, + const std::vector& outputColumns) { + query_ = query; + outputColumns_ = outputColumns; + errorCode_ = preprocessQuery(); +} + +uint64_t ClpCursor::fetchNext( + uint64_t numRows, + const std::shared_ptr>& filteredRowIndices) { + if (ErrorCode::Success != errorCode_) { + return 0; + } + + if (false == currentArchiveLoaded_) { + errorCode_ = loadArchive(); + if (ErrorCode::Success != errorCode_) { + return 0; + } + + archiveReader_->open_packed_streams(); + currentArchiveLoaded_ = true; + queryRunner_ = std::make_shared( + schemaMatch_, expr_, archiveReader_, false, projection_); + queryRunner_->global_init(); + } + + while (currentSchemaIndex_ < matchedSchemas_.size()) { + if (false == currentSchemaTableLoaded_) { + currentSchemaId_ = matchedSchemas_[currentSchemaIndex_]; + if (EvaluatedValue::False == + queryRunner_->schema_init(currentSchemaId_)) { + currentSchemaIndex_ += 1; + currentSchemaTableLoaded_ = false; + errorCode_ = ErrorCode::DictionaryNotFound; + continue; + } + + auto& reader = + archiveReader_->read_schema_table(currentSchemaId_, false, false); + reader.initialize_filter_with_column_map(queryRunner_.get()); + + errorCode_ = ErrorCode::Success; + currentSchemaTableLoaded_ = true; + } + + auto rowsScanned = queryRunner_->fetchNext(numRows, filteredRowIndices); + if (false == filteredRowIndices->empty()) { + return rowsScanned; + } + + currentSchemaIndex_ += 1; + currentSchemaTableLoaded_ = false; + } + + return 0; +} + +const std::vector& ClpCursor::getProjectedColumns() + const { + if (queryRunner_) { + return queryRunner_->getProjectedColumns(); + } + static std::vector const kEmpty; + return kEmpty; +} + +ErrorCode ClpCursor::preprocessQuery() { + auto queryStream = std::istringstream(query_); + expr_ = kql::parse_kql_expression(queryStream); + if (nullptr == expr_) { + VLOG(2) << "Failed to parse query '" << query_ << "'"; + return ErrorCode::InvalidQuerySyntax; + } + + if (std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + OrOfAndForm standardizePass; + if (expr_ = standardizePass.run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + NarrowTypes narrowPass; + if (expr_ = narrowPass.run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + ConvertToExists convertPass; + if (expr_ = convertPass.run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "Query '" << query_ << "' is logically false"; + return ErrorCode::LogicalError; + } + + return ErrorCode::Success; +} + +ErrorCode ClpCursor::loadArchive() { + auto networkAuthOption = inputSource_ == InputSource::Filesystem + ? NetworkAuthOption{.method = AuthMethod::None} + : NetworkAuthOption{.method = AuthMethod::S3PresignedUrlV4}; + + try { + archiveReader_->open( + get_path_object_for_raw_path(archivePath_), networkAuthOption); + } catch (std::exception& e) { + VLOG(2) << "Failed to open archive file: " << e.what(); + return ErrorCode::InternalError; + } + + auto timestampDict = archiveReader_->get_timestamp_dictionary(); + auto schemaTree = archiveReader_->get_schema_tree(); + auto schemaMap = archiveReader_->get_schema_map(); + + EvaluateTimestampIndex timestampIndex(timestampDict); + if (clp_s::EvaluatedValue::False == timestampIndex.run(expr_)) { + VLOG(2) << "No matching timestamp ranges for query '" << query_ << "'"; + return ErrorCode::InvalidTimestampRange; + } + + schemaMatch_ = std::make_shared(schemaTree, schemaMap); + if (expr_ = schemaMatch_->run(expr_); + std::dynamic_pointer_cast(expr_)) { + VLOG(2) << "No matching schemas for query '" << query_ << "'"; + return ErrorCode::SchemaNotFound; + } + + projection_ = std::make_shared( + outputColumns_.empty() ? ReturnAllColumns : ReturnSelectedColumns); + try { + for (auto const& column : outputColumns_) { + std::vector descriptorTokens; + std::string descriptorNamespace; + if (false == + tokenize_column_descriptor( + column.name, descriptorTokens, descriptorNamespace)) { + VLOG(2) << "Can not tokenize invalid column: '" << column.name << "'"; + return ErrorCode::InternalError; + } + + auto columnDescriptor = ColumnDescriptor::create_from_escaped_tokens( + descriptorTokens, descriptorNamespace); + switch (column.type) { + case ColumnType::String: + columnDescriptor->set_matching_types( + LiteralType::ClpStringT | LiteralType::VarStringT | + LiteralType::EpochDateT); + break; + case ColumnType::Integer: + columnDescriptor->set_matching_types(LiteralType::IntegerT); + break; + case ColumnType::Float: + columnDescriptor->set_matching_types(LiteralType::FloatT); + break; + case ColumnType::Boolean: + columnDescriptor->set_matching_types(LiteralType::BooleanT); + break; + case ColumnType::Array: + columnDescriptor->set_matching_types(LiteralType::ArrayT); + break; + default: + break; + } + + projection_->add_column(columnDescriptor); + } + } catch (TraceableException& e) { + VLOG(2) << e.what(); + return ErrorCode::InternalError; + } + projection_->resolve_columns(schemaTree); + archiveReader_->set_projection(projection_); + + archiveReader_->read_metadata(); + + matchedSchemas_.clear(); + for (auto schemaId : archiveReader_->get_schema_ids()) { + if (schemaMatch_->schema_matched(schemaId)) { + matchedSchemas_.push_back(schemaId); + } + } + + if (matchedSchemas_.empty()) { + return ErrorCode::SchemaNotFound; + } + + EvaluateTimestampIndex timestamp_index(timestampDict); + if (EvaluatedValue::False == timestamp_index.run(expr_)) { + VLOG(2) << "No matching timestamp ranges for query '" << query_ << "'"; + return ErrorCode::InvalidTimestampRange; + } + + archiveReader_->read_variable_dictionary(); + archiveReader_->read_log_type_dictionary(); + archiveReader_->read_array_dictionary(); + + currentSchemaIndex_ = 0; + currentSchemaTableLoaded_ = false; + return ErrorCode::Success; +} + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpCursor.h b/velox/connectors/clp/search_lib/ClpCursor.h new file mode 100644 index 00000000000..43db49c7841 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpCursor.h @@ -0,0 +1,125 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include + +#include "velox/connectors/clp/search_lib/ClpQueryRunner.h" + +namespace clp_s { +enum class InputSource : uint8_t; +class ArchiveReader; +class BaseColumnReader; +} // namespace clp_s + +namespace clp_s::search { +class Projection; +class SchemaMatch; +} // namespace clp_s::search + +namespace clp_s::search::ast { +class Expression; +} // namespace clp_s::search::ast + +namespace facebook::velox::connector::clp::search_lib { + +enum class ErrorCode { + DictionaryNotFound, + InternalError, + InvalidQuerySyntax, + InvalidTimestampRange, + LogicalError, + QueryNotInitialized, + SchemaNotFound, + Success +}; + +enum class ColumnType { String, Integer, Float, Array, Boolean, Unknown = -1 }; + +struct Field { + ColumnType type; + std::string name; +}; + +/// A query execution interface that manages the lifecycle of a query on a CLP-S +/// archive, including parsing and validating the query, loading the relevant +/// schemas and archives, applying filters, and iterating over the results. It +/// abstracts away the low-level details of archive access and schema matching +/// while supporting projection and batch-oriented retrieval of filtered rows. +class ClpCursor { + public: + explicit ClpCursor(clp_s::InputSource inputSource, std::string archivePath); + ~ClpCursor(); + + /// Executes a query. This function parses, validates, and prepares the given + /// query for execution. + /// + /// @param query The KQL query to execute. + /// @param outputColumns A vector specifying the columns to be included in the + /// query result. + void executeQuery( + const std::string& query, + const std::vector& outputColumns); + + /// Fetches the next set of rows from the cursor. If the archive and schema + /// are not yet loaded, this function will perform the necessary loading. + /// + /// @param numRows The maximum number of rows to fetch. + /// @param filteredRowIndices A vector of row indices that match the filter. + /// @return The number of rows scanned. + uint64_t fetchNext( + uint64_t numRows, + const std::shared_ptr>& filteredRowIndices); + + /// Retrieves the projected columns. + /// + /// @return A vector of BaseColumnReader pointers representing the projected + /// columns. + const std::vector& getProjectedColumns() const; + + private: + /// Preprocesses the query, performing parsing, validation, and optimization. + /// + /// @return The error code. + ErrorCode preprocessQuery(); + + /// Loads the archive at the current index. + /// + /// @return The error code. + ErrorCode loadArchive(); + + ErrorCode errorCode_; + + clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem}; + std::string archivePath_; + std::string query_; + std::vector outputColumns_; + std::vector matchedSchemas_; + size_t currentSchemaIndex_{0}; + int32_t currentSchemaId_{-1}; + bool currentSchemaTableLoaded_{false}; + bool currentArchiveLoaded_{false}; + + std::shared_ptr expr_; + std::shared_ptr schemaMatch_; + std::shared_ptr queryRunner_; + std::shared_ptr projection_; + std::shared_ptr archiveReader_; +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpQueryRunner.cpp b/velox/connectors/clp/search_lib/ClpQueryRunner.cpp new file mode 100644 index 00000000000..a1c9032ebd5 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpQueryRunner.cpp @@ -0,0 +1,81 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "velox/connectors/clp/search_lib/ClpQueryRunner.h" + +using namespace clp_s; +using namespace clp_s::search; +using namespace clp_s::search::clp_search; + +namespace facebook::velox::connector::clp::search_lib { + +void ClpQueryRunner::init( + clp_s::SchemaReader* schemaReader, + std::unordered_map const& columnMap) { + numMessages_ = schemaReader->get_num_messages(); + curMessage_ = 0; + clear_readers(); + + projectedColumns_.clear(); + auto matchingNodesList = projection_->get_ordered_matching_nodes(); + for (const auto& nodeIds : matchingNodesList) { + if (nodeIds.empty()) { + projectedColumns_.push_back(nullptr); + continue; + } + + // Try to find a matching column in columnMap + bool foundReader = false; + for (const auto nodeId : nodeIds) { + auto columnIt = columnMap.find(nodeId); + if (columnIt != columnMap.end()) { + projectedColumns_.push_back(columnIt->second); + foundReader = true; + break; + } + } + + if (!foundReader) { + projectedColumns_.push_back(nullptr); + } + } + + for (const auto& [columnId, columnReader] : columnMap) { + initialize_reader(columnId, columnReader); + } +} + +uint64_t ClpQueryRunner::fetchNext( + uint64_t numRows, + const std::shared_ptr>& filteredRowIndices) { + size_t rowsfiltered{0}; + size_t rowsScanned{0}; + while (curMessage_ < numMessages_) { + if (filter(curMessage_)) { + filteredRowIndices->emplace_back(curMessage_); + rowsfiltered += 1; + } + + curMessage_ += 1; + rowsScanned += 1; + if (rowsfiltered >= numRows) { + break; + } + } + return rowsScanned; +} + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpQueryRunner.h b/velox/connectors/clp/search_lib/ClpQueryRunner.h new file mode 100644 index 00000000000..97c558e096e --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpQueryRunner.h @@ -0,0 +1,90 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "clp_s/search/QueryRunner.hpp" + +namespace clp_s { +class SchemaReader; +class SchemaTree; +class ArchiveReader; +class BaseColumnReader; +} // namespace clp_s + +namespace clp_s::search { +class SchemaMatch; +class Projection; +} // namespace clp_s::search + +namespace clp_s::search::ast { +class Expression; +} // namespace clp_s::search::ast + +namespace facebook::velox::connector::clp::search_lib { + +/// Extends the generic QueryRunner to support column projection and row +/// filtering over CLP-S archives. It is used by the Velox-CLP connector to +/// efficiently identify matching rows and project relevant columns, which are +/// then consumed by the ClpVectorLoader. +class ClpQueryRunner : public clp_s::search::QueryRunner { + public: + ClpQueryRunner( + const std::shared_ptr& match, + const std::shared_ptr& expr, + const std::shared_ptr& archiveReader, + bool ignoreCase, + const std::shared_ptr& projection) + : clp_s::search::QueryRunner(match, expr, archiveReader, ignoreCase), + projection_(projection) {} + + /// Initializes the filter with schema information and column readers. + /// + /// @param schemaReader A pointer to the SchemaReader. + /// @param columnMap An unordered map associating column IDs with + /// BaseColumnReader pointers. + void init( + clp_s::SchemaReader* schemaReader, + std::unordered_map const& columnMap) + override; + + /// Fetches the next set of rows from the cursor. + /// + /// @param numRows The maximum number of rows to fetch. + /// @param filteredRowIndices A vector to store the row indices that match the + /// filter. + /// @return The number of rows scanned. + uint64_t fetchNext( + uint64_t numRows, + const std::shared_ptr>& filteredRowIndices); + + /// @return A reference to the vector of BaseColumnReader pointers that + /// represent the columns involved in the scanning operation. + const std::vector& getProjectedColumns() { + return projectedColumns_; + } + + private: + std::shared_ptr expr_; + std::shared_ptr schemaTree_; + std::shared_ptr projection_; + std::vector projectedColumns_; + + uint64_t curMessage_{}; + uint64_t numMessages_{}; +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp new file mode 100644 index 00000000000..be73e842d73 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp @@ -0,0 +1,164 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include "clp_s/ColumnReader.hpp" + +#include "velox/connectors/clp/search_lib/ClpVectorLoader.h" +#include "velox/vector/ComplexVector.h" +#include "velox/vector/FlatVector.h" + +namespace facebook::velox::connector::clp::search_lib { + +ClpVectorLoader::ClpVectorLoader( + clp_s::BaseColumnReader* columnReader, + ColumnType nodeType, + std::shared_ptr> filteredRowIndices) + : columnReader_(columnReader), + nodeType_(nodeType), + filteredRowIndices_(std::move(filteredRowIndices)) {} + +template +void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) { + if (columnReader_ == nullptr) { + for (int vectorIndex : rows) { + vector->setNull(vectorIndex, true); + } + return; + } + + for (int vectorIndex : rows) { + auto messageIndex = (*filteredRowIndices_)[vectorIndex]; + + if constexpr (std::is_same_v) { + auto string_value = + std::get(columnReader_->extract_value(messageIndex)); + vector->set(vectorIndex, StringView(string_value)); + } else { + vector->set( + vectorIndex, std::get(columnReader_->extract_value(messageIndex))); + } + + vector->setNull(vectorIndex, false); + } +} + +void ClpVectorLoader::loadInternal( + RowSet rows, + ValueHook* hook, + vector_size_t resultSize, + VectorPtr* result) { + VELOX_CHECK_NOT_NULL(result, "result vector must not be null"); + + auto vector = *result; + switch (nodeType_) { + case ColumnType::Integer: { + auto intVector = vector->asFlatVector(); + populateData(rows, intVector); + break; + } + case ColumnType::Float: { + auto floatVector = vector->asFlatVector(); + populateData(rows, floatVector); + break; + } + case ColumnType::Boolean: { + auto boolVector = vector->asFlatVector(); + populateData(rows, boolVector); + break; + } + case ColumnType::String: { + auto stringVector = vector->asFlatVector(); + populateData(rows, stringVector); + break; + } + case ColumnType::Array: { + if (columnReader_ == nullptr) { + for (int vectorIndex : rows) { + vector->setNull(vectorIndex, true); + } + return; + } + auto arrayVector = std::dynamic_pointer_cast(vector); + auto elements = arrayVector->elements()->asFlatVector(); + vector_size_t elementIndex = 0; + + for (int vectorIndex : rows) { + auto messageIndex = (*filteredRowIndices_)[vectorIndex]; + + auto jsonString = + std::get(columnReader_->extract_value(messageIndex)); + + simdjson::padded_string padded(jsonString); + simdjson::ondemand::document doc; + try { + doc = arrayParser_->iterate(padded); + } catch (const simdjson::simdjson_error& e) { + VELOX_FAIL("JSON parse error at row {}: {}", vectorIndex, e.what()); + } + + simdjson::ondemand::array array; + try { + array = doc.get_array(); + } catch (const simdjson::simdjson_error& e) { + VELOX_FAIL( + "Expected JSON array at row {}: {}", vectorIndex, e.what()); + } + + std::vector arrayElements; + for (auto arrayElement : array) { + arrayElements.emplace_back( + simdjson::to_json_string(arrayElement).value()); + } + + if (elementIndex + arrayElements.size() > elements->size()) { + size_t newSize = std::max( + elementIndex + arrayElements.size(), + static_cast(elements->size()) * 2); + elements->resize(newSize); + } + + arrayVector->setOffsetAndSize( + vectorIndex, elementIndex, arrayElements.size()); + for (auto& arrayElement : arrayElements) { + elements->set(elementIndex++, StringView(arrayElement)); + } + + arrayVector->setNull(vectorIndex, false); + } + break; + } + default: + VELOX_FAIL("Unsupported column type"); + } +} + +// Explicit template instantiations for linker +template void ClpVectorLoader::populateData( + RowSet rows, + FlatVector* vector); +template void ClpVectorLoader::populateData( + RowSet rows, + FlatVector* vector); +template void ClpVectorLoader::populateData( + RowSet rows, + FlatVector* vector); +template void ClpVectorLoader::populateData( + RowSet rows, + FlatVector* vector); + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.h b/velox/connectors/clp/search_lib/ClpVectorLoader.h new file mode 100644 index 00000000000..ab57b52dfd9 --- /dev/null +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.h @@ -0,0 +1,56 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/vector/LazyVector.h" + +namespace clp_s { +class BaseColumnReader; +} // namespace clp_s + +namespace facebook::velox::connector::clp::search_lib { + +/// A custom Velox VectorLoader that populates Velox vectors from a CLP-based +/// column reader. It supports various column types including integers, floats, +/// booleans, strings, and arrays of strings. +class ClpVectorLoader : public VectorLoader { + public: + ClpVectorLoader( + clp_s::BaseColumnReader* columnReader, + ColumnType nodeType, + std::shared_ptr> filteredRowIndices); + + private: + void loadInternal( + RowSet rows, + ValueHook* hook, + vector_size_t resultSize, + VectorPtr* result) override; + + template + void populateData(RowSet rows, VectorPtr vector); + + clp_s::BaseColumnReader* columnReader_; + ColumnType nodeType_; + std::shared_ptr> filteredRowIndices_; + + inline static thread_local std::unique_ptr + arrayParser_ = std::make_unique(); +}; + +} // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/tests/CMakeLists.txt b/velox/connectors/clp/tests/CMakeLists.txt new file mode 100644 index 00000000000..4e53803374b --- /dev/null +++ b/velox/connectors/clp/tests/CMakeLists.txt @@ -0,0 +1,27 @@ +# Copyright (c) Facebook, Inc. and its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +add_executable(velox_clp_connector_test ClpConnectorTest.cpp) + +add_test(velox_clp_connector_test velox_clp_connector_test) + +target_link_libraries( + velox_clp_connector_test + velox_clp_connector + velox_exec_test_lib + velox_vector_test_lib + GTest::gtest + GTest::gtest_main) + +file(COPY ${CMAKE_CURRENT_SOURCE_DIR}/examples + DESTINATION ${CMAKE_CURRENT_BINARY_DIR}) diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp new file mode 100644 index 00000000000..bb351984b64 --- /dev/null +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -0,0 +1,311 @@ +/* + * Copyright (c) Facebook, Inc. and its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include + +#include "velox/common/base/Fs.h" +#include "velox/common/base/tests/GTestUtils.h" +#include "velox/connectors/clp/ClpColumnHandle.h" +#include "velox/connectors/clp/ClpConnector.h" +#include "velox/connectors/clp/ClpConnectorSplit.h" +#include "velox/connectors/clp/ClpTableHandle.h" +#include "velox/exec/tests/utils/AssertQueryBuilder.h" +#include "velox/exec/tests/utils/OperatorTestBase.h" +#include "velox/exec/tests/utils/PlanBuilder.h" + +namespace { + +using namespace facebook::velox; +using namespace facebook::velox::connector::clp; + +using facebook::velox::exec::test::PlanBuilder; + +class ClpConnectorTest : public exec::test::OperatorTestBase { + public: + const std::string kClpConnectorId = "test-clp"; + + void SetUp() override { + OperatorTestBase::SetUp(); + connector::registerConnectorFactory( + std::make_shared()); + auto clpConnector = + connector::getConnectorFactory( + connector::clp::ClpConnectorFactory::kClpConnectorName) + ->newConnector( + kClpConnectorId, + std::make_shared( + std::unordered_map{ + {"clp.split-source", "local"}})); + connector::registerConnector(clpConnector); + } + + void TearDown() override { + connector::unregisterConnector(kClpConnectorId); + connector::unregisterConnectorFactory( + connector::clp::ClpConnectorFactory::kClpConnectorName); + OperatorTestBase::TearDown(); + } + + exec::Split makeClpSplit(const std::string& splitPath) { + return exec::Split( + std::make_shared(kClpConnectorId, splitPath)); + } + + RowVectorPtr getResults( + const core::PlanNodePtr& planNode, + std::vector&& splits) { + return exec::test::AssertQueryBuilder(planNode) + .splits(std::move(splits)) + .copyResults(pool()); + } + + static std::string getExampleFilePath(const std::string& filePath) { + std::string current_path = fs::current_path().string(); + return current_path + "/examples/" + filePath; + } +}; + +TEST_F(ClpConnectorTest, test1NoPushdown) { + auto plan = PlanBuilder() + .startTableScan() + .outputType( + ROW({"requestId", "userId", "method"}, + {VARCHAR(), VARCHAR(), VARCHAR()})) + .tableHandle(std::make_shared( + kClpConnectorId, "test_1", nullptr)) + .assignments({ + {"requestId", + std::make_shared( + "requestId", "requestId", VARCHAR(), true)}, + {"userId", + std::make_shared( + "userId", "userId", VARCHAR(), true)}, + {"method", + std::make_shared( + "method", "method", VARCHAR(), true)}, + }) + .endTableScan() + .filter("method = 'GET'") + .planNode(); + + auto output = + getResults(plan, {makeClpSplit(getExampleFilePath("test_1.clps"))}); + auto expected = makeRowVector( + {// requestId + makeFlatVector( + {"req-100", "req-105", "req-107", "req-109", "req-102"}), + // userId + makeNullableFlatVector( + {"user201", "user204", "user202", "user203", std::nullopt}), + // method + makeFlatVector({ + "GET", + "GET", + "GET", + "GET", + "GET", + })}); + test::assertEqualVectors(expected, output); +} + +TEST_F(ClpConnectorTest, test1Pushdown) { + auto plan = PlanBuilder() + .startTableScan() + .outputType( + ROW({"requestId", "userId", "path"}, + {VARCHAR(), VARCHAR(), VARCHAR()})) + .tableHandle(std::make_shared( + kClpConnectorId, + "test_1", + std::make_shared( + "method: \"POST\" AND status: 200"))) + .assignments({ + {"requestId", + std::make_shared( + "requestId", "requestId", VARCHAR(), true)}, + {"userId", + std::make_shared( + "userId", "userId", VARCHAR(), true)}, + {"path", + std::make_shared( + "path", "path", VARCHAR(), true)}, + }) + .endTableScan() + .planNode(); + + auto output = + getResults(plan, {makeClpSplit(getExampleFilePath("test_1.clps"))}); + auto expected = + makeRowVector({// requestId + makeFlatVector({"req-106"}), + // userId + makeNullableFlatVector({std::nullopt}), + // path + makeFlatVector({"/auth/login"})}); + test::assertEqualVectors(expected, output); +} + +TEST_F(ClpConnectorTest, test2NoPushdown) { + auto plan = + PlanBuilder(pool_.get()) + .startTableScan() + .outputType( + ROW({"timestamp", "event"}, + {VARCHAR(), + ROW({"type", "subtype", "severity"}, + {VARCHAR(), VARCHAR(), VARCHAR()})})) + .tableHandle(std::make_shared( + kClpConnectorId, "test_2", nullptr)) + .assignments( + {{"timestamp", + std::make_shared( + "timestamp", "timestamp", VARCHAR(), true)}, + {"event", + std::make_shared( + "event", + "event", + ROW({"type", "subtype", "severity"}, + {VARCHAR(), VARCHAR(), VARCHAR()}), + true)}}) + .endTableScan() + .filter( + "event.severity IN ('WARNING', 'ERROR') AND " + "((event.type = 'network' AND event.subtype = 'connection') OR " + "(event.type = 'storage' AND event.subtype LIKE 'disk_usage%'))") + .planNode(); + + auto output = + getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))}); + auto expected = + makeRowVector({// timestamp + makeFlatVector({"2025-04-30T08:50:05Z"}), + // event + makeRowVector({ + // event.type + makeFlatVector({"storage"}), + // event.subtype + makeFlatVector({"disk_usage"}), + // event.severity + makeFlatVector({"WARNING"}), + })}); + test::assertEqualVectors(expected, output); +} + +TEST_F(ClpConnectorTest, test2Pushdown) { + auto plan = + PlanBuilder() + .startTableScan() + .outputType( + ROW({"timestamp", "event"}, + {VARCHAR(), + ROW({"type", "subtype", "severity"}, + {VARCHAR(), VARCHAR(), VARCHAR()})})) + .tableHandle(std::make_shared( + kClpConnectorId, + "test_2", + std::make_shared( + "(event.severity: \"WARNING\" OR event.severity: \"ERROR\") AND " + "((event.type: \"network\" AND event.subtype: \"connection\") OR " + "(event.type: \"storage\" AND event.subtype: \"disk*\"))"))) + .assignments( + {{"timestamp", + std::make_shared( + "timestamp", "timestamp", VARCHAR(), true)}, + {"event", + std::make_shared( + "event", + "event", + ROW({"type", "subtype", "severity"}, + {VARCHAR(), VARCHAR(), VARCHAR()}), + true)}}) + .endTableScan() + .planNode(); + + auto output = + getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))}); + auto expected = + makeRowVector({// timestamp + makeFlatVector({"2025-04-30T08:50:05Z"}), + // event + makeRowVector({ + // event.type + makeFlatVector({"storage"}), + // event.subtype + makeFlatVector({"disk_usage"}), + // event.severity + makeFlatVector({"WARNING"}), + })}); + test::assertEqualVectors(expected, output); +} + +TEST_F(ClpConnectorTest, test2Hybrid) { + auto plan = + PlanBuilder(pool_.get()) + .startTableScan() + .outputType( + ROW({"timestamp", "event"}, + {VARCHAR(), + ROW({"type", "subtype", "severity", "tags"}, + {VARCHAR(), VARCHAR(), VARCHAR(), ARRAY(VARCHAR())})})) + .tableHandle(std::make_shared( + kClpConnectorId, + "test_2", + std::make_shared( + "((event.type: \"network\" AND event.subtype: \"connection\") OR " + "(event.type: \"storage\" AND event.subtype: \"disk*\"))"))) + .assignments( + {{"timestamp", + std::make_shared( + "timestamp", "timestamp", VARCHAR(), true)}, + {"event", + std::make_shared( + "event", + "event", + ROW({"type", "subtype", "severity", "tags"}, + {VARCHAR(), VARCHAR(), VARCHAR(), ARRAY(VARCHAR())}), + true)}}) + .endTableScan() + .filter("upper(event.severity) IN ('WARNING', 'ERROR')") + .planNode(); + + auto output = + getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))}); + auto expected = makeRowVector( + {// timestamp + makeFlatVector({"2025-04-30T08:50:05Z"}), + // event + makeRowVector({// event.type + makeFlatVector({"storage"}), + // event.subtype + makeFlatVector({"disk_usage"}), + // event.severity + makeFlatVector({"WARNING"}), + // event.tags + makeArrayVector( + {{"\"filesystem\"", "\"monitoring\""}})}) + + }); + test::assertEqualVectors(expected, output); +} + +} // namespace + +int main(int argc, char** argv) { + testing::InitGoogleTest(&argc, argv); + folly::Init init{&argc, &argv, false}; + return RUN_ALL_TESTS(); +} diff --git a/velox/connectors/clp/tests/examples/test_1.clps b/velox/connectors/clp/tests/examples/test_1.clps new file mode 100644 index 00000000000..9538d88c997 Binary files /dev/null and b/velox/connectors/clp/tests/examples/test_1.clps differ diff --git a/velox/connectors/clp/tests/examples/test_1.ndjson b/velox/connectors/clp/tests/examples/test_1.ndjson new file mode 100644 index 00000000000..aca627d5fad --- /dev/null +++ b/velox/connectors/clp/tests/examples/test_1.ndjson @@ -0,0 +1,10 @@ +{"timestamp": "2025-04-30T08:45:00Z", "requestId": "req-100", "userId": "user201", "method": "GET", "path": "/api/users/1", "responseTimeMs": 25, "status": 200} +{"timestamp": "2025-04-30T08:45:05Z", "requestId": "req-101", "userId": "user202", "method": "POST", "path": "/api/orders", "responseTimeMs": 110, "status": 201} +{"timestamp": "2025-04-30T08:45:10Z", "requestId": "req-102", "userId": null, "method": "GET", "path": "/public/products", "responseTimeMs": 18, "status": 200} +{"timestamp": "2025-04-30T08:45:15Z", "requestId": "req-103", "userId": "user203", "method": "PUT", "path": "/api/items/5", "responseTimeMs": 95, "status": 200} +{"timestamp": "2025-04-30T08:45:20Z", "requestId": "req-104", "userId": "user201", "method": "DELETE", "path": "/api/notifications/3", "responseTimeMs": 32, "status": 204} +{"timestamp": "2025-04-30T08:45:25Z", "requestId": "req-105", "userId": "user204", "method": "GET", "path": "/api/dashboard", "responseTimeMs": 155, "status": 200} +{"timestamp": "2025-04-30T08:45:30Z", "requestId": "req-106", "userId": null, "method": "POST", "path": "/auth/login", "responseTimeMs": 68, "status": 200} +{"timestamp": "2025-04-30T08:45:35Z", "requestId": "req-107", "userId": "user202", "method": "GET", "path": "/api/users/2/details", "responseTimeMs": 41, "status": 200} +{"timestamp": "2025-04-30T08:45:40Z", "requestId": "req-108", "userId": "user205", "method": "PATCH", "path": "/api/settings", "responseTimeMs": 128, "status": 200} +{"timestamp": "2025-04-30T08:45:45Z", "requestId": "req-109", "userId": "user203", "method": "GET", "path": "/api/products?category=books", "responseTimeMs": 88, "status": 200} diff --git a/velox/connectors/clp/tests/examples/test_2.clps b/velox/connectors/clp/tests/examples/test_2.clps new file mode 100644 index 00000000000..ad773162d3a Binary files /dev/null and b/velox/connectors/clp/tests/examples/test_2.clps differ diff --git a/velox/connectors/clp/tests/examples/test_2.ndjson b/velox/connectors/clp/tests/examples/test_2.ndjson new file mode 100644 index 00000000000..8bccc2d3168 --- /dev/null +++ b/velox/connectors/clp/tests/examples/test_2.ndjson @@ -0,0 +1,10 @@ +{"timestamp": "2025-04-30T08:50:00Z", "event": {"type": "network", "subtype": "connection", "severity": "INFO", "tags": ["inbound", "tcp"], "details": {"source": {"ip": "192.168.3.10"}, "destination": {"ip": "10.0.1.5", "port": 80}}}} +{"timestamp": "2025-04-30T08:50:05Z", "event": {"type": "storage", "subtype": "disk_usage", "severity": "WARNING", "tags": ["filesystem", "monitoring"], "details": {"mount": "/var/log", "usage": {"percent": 92}}}} +{"timestamp": "2025-04-30T08:50:10Z", "event": {"type": "process", "subtype": "start", "severity": "INFO", "tags": ["system", "init"], "details": {"name": "cron", "id": 1234}}} +{"timestamp": "2025-04-30T08:50:15Z", "event": {"type": "authentication", "subtype": "failure", "severity": "ERROR", "tags": ["login", "security"], "user": {"name": "guest", "method": "ssh"}, "details": {"reason": "invalid password", "ip": "203.0.113.20"}}} +{"timestamp": "2025-04-30T08:50:20Z", "event": {"type": "service", "subtype": "status", "severity": "INFO", "tags": ["daemon", "http"], "details": {"name": "httpd", "state": "running", "uptime": {"seconds": 3600}}}} +{"timestamp": "2025-04-30T08:50:25Z", "event": {"type": "security", "subtype": "alert", "severity": "HIGH", "tags": ["threat", "sql"], "details": {"rule": {"id": "SQLI-001"}, "target": "/api/data", "payload": "SELECT * FROM users;"}}} +{"timestamp": "2025-04-30T08:50:30Z", "event": {"type": "memory", "subtype": "usage", "severity": "NORMAL", "tags": ["resources"], "details": {"total": {"gb": 16}, "used": {"gb": 8.5}, "free": {"gb": 7.5}}}} +{"timestamp": "2025-04-30T08:50:35Z", "event": {"type": "configuration", "subtype": "change", "severity": "INFO", "tags": ["admin", "settings"], "user": {"name": "admin"}, "details": {"component": "database", "setting": {"name": "timeout", "oldValue": 30, "newValue": 60}}}} +{"timestamp": "2025-04-30T08:50:40Z", "event": {"type": "file", "subtype": "access", "severity": "WARNING", "tags": ["file", "privileged"], "user": {"name": "user101"}, "details": {"operation": "read", "path": "/etc/shadow", "permissions": "rw-------", "flags": ["sensitive", "audit"]}}} +{"timestamp": "2025-04-30T08:50:45Z", "event": {"type": "backup", "subtype": "status", "severity": "INFO", "tags": ["backup", "daily"], "details": {"job": {"name": "daily_backup", "status": "completed"}, "duration": {"seconds": 1200}, "size": {"gb": 25.3}}}} diff --git a/velox/docs/configs.rst b/velox/docs/configs.rst index 9c19b600e06..e7864ac2fd2 100644 --- a/velox/docs/configs.rst +++ b/velox/docs/configs.rst @@ -900,6 +900,21 @@ These semantics are similar to the `Apache Hadoop-Aws module /oauth2/token`. +CLP Connector +----------------------------- +.. list-table:: + :widths: 20 10 10 70 + :header-rows: 1 + + * - Property Name + - Type + - Default Value + - Description + * - clp.storage-type + - string + - FS + - The storage type of CLP archives, **Allowed values:** ``FS``, ``S3``. + Presto-specific Configuration ----------------------------- .. list-table:: diff --git a/velox/docs/develop/connectors.rst b/velox/docs/develop/connectors.rst index 2d7ac8a011f..21a486e84a5 100644 --- a/velox/docs/develop/connectors.rst +++ b/velox/docs/develop/connectors.rst @@ -29,8 +29,8 @@ Connector Interface * - Connector Factory - Enables creating instances of a particular connector. -Velox provides Hive and TPC-H Connectors out of the box. -Let's see how the above connector interfaces are implemented in the Hive Connector in detail below. +Velox provides Hive, CLP, and TPC-H Connectors out of the box. +Let's examine the implementation details of both the Hive and CLP Connectors as examples. Hive Connector -------------- @@ -121,3 +121,29 @@ This is the behavior when the proxy settings are enabled: 4. The no_proxy/NO_PROXY list is comma separated. 5. Use . or \*. to indicate domain suffix matching, e.g. `.foobar.com` will match `test.foobar.com` or `foo.foobar.com`. + +CLP Connector +------------- +The CLP Connector is used to read CLP archives stored on a local file system or S3. It implements similar +interfaces as the Hive Connector except for the ``DataSink`` interface. Here we only describe the ``DataSource`` +interface and the ``ConnectorSplit`` interface implementation since `Connector` and ``ConnectorFactory`` are +similar to the Hive Connector. + +ClpConnectorSplit +~~~~~~~~~~~~~~~~~ +``ClpConnectorSplit`` describes a data chunk using ``path``, which is the path to the archive file. + +ClpDataSource +~~~~~~~~~~~~~ +``ClpDataSource`` implements the ``addSplit`` API that consumes a ``ClpConnectorSplit`` and ``next`` API that +processes the split and returns a batch of rows. + +During initialization, it records the KQL query and archive source (S3 or local). It then iterates through +each output column, accessing its handle to get its type and original name. For row types, it recursively +traverses the nested structure to process each field; for non-row types, it directly maps the Velox column +type to a CLP column type. + +When a split is added, a ``ClpCursor`` is created with the archive path and input source. The query is parsed +and simplified into an AST. On ``next``, the cursor finds matching row indices and, if any exist, +``ClpDataSource`` recursively creates a row vector composed of lazy vectors, which use CLP column readers to +decode and load data as needed during execution.