1. remove extra # in cmake files

2. fix a bug in FindANTLR.cmake
3. download antlr jar to build directory instead of the source directory
4. add StorageType to ClpTableHandle
5. use consistent types for filteredRows across different methods
6. remove archiveSource_ from ClpConfig and config docs, use storageType_ in ClpDataSource
7. minor improvements in logs and method signatures
8. rename splitPath to path in ClpConnectorSplit

Author: wraymo

CMake/ExternalAntlr4Cpp.cmake

@@ -99,7 +99,7 @@ if(ANTLR4_ZIP_REPOSITORY)
       -DDISABLE_WARNINGS:BOOL=ON
       # -DCMAKE_CXX_STANDARD:STRING=17 # if desired, compile the runtime with a
       # different C++ standard -DCMAKE_CXX_STANDARD:STRING=${CMAKE_CXX_STANDARD}
-      # # alternatively, compile the runtime with the same C++ standard as the
+      # alternatively, compile the runtime with the same C++ standard as the
       # outer project
     INSTALL_COMMAND ""
     EXCLUDE_FROM_ALL 1)
@@ -120,7 +120,7 @@ else()
       -DDISABLE_WARNINGS:BOOL=ON
       # -DCMAKE_CXX_STANDARD:STRING=17 # if desired, compile the runtime with a
       # different C++ standard -DCMAKE_CXX_STANDARD:STRING=${CMAKE_CXX_STANDARD}
-      # # alternatively, compile the runtime with the same C++ standard as the
+      # alternatively, compile the runtime with the same C++ standard as the
       # outer project
     INSTALL_COMMAND ""
     EXCLUDE_FROM_ALL 1)

CMake/FindANTLR.cmake

@@ -21,10 +21,13 @@ add_definitions(-DANTLR4CPP_STATIC)
 
 # Define the JAR name and path
 set(ANTLR_JAR_NAME antlr-${ANTLR4_TAG}-complete.jar)
-set(ANTLR_EXECUTABLE ${PROJECT_SOURCE_DIR}/third_party/antlr/${ANTLR_JAR_NAME})
 
-# Ensure the output directory exists
-file(MAKE_DIRECTORY ${PROJECT_SOURCE_DIR}/third_party/antlr)
+# Set the download directory
+set(ANTLR_DOWNLOAD_DIR ${CMAKE_BINARY_DIR}/antlr)
+file(MAKE_DIRECTORY ${ANTLR_DOWNLOAD_DIR})
+
+# Define the full path to the ANTLR JAR
+set(ANTLR_EXECUTABLE ${ANTLR_DOWNLOAD_DIR}/${ANTLR_JAR_NAME})
 
 # Download the ANTLR JAR if it does not exist
 if(NOT EXISTS ${ANTLR_EXECUTABLE})
@@ -84,7 +87,6 @@ if(ANTLR_EXECUTABLE AND Java_JAVA_EXECUTABLE)
     if(ANTLR_TARGET_OUTPUT_DIRECTORY)
       set(ANTLR_${Name}_OUTPUT_DIR ${ANTLR_TARGET_OUTPUT_DIRECTORY})
     else()
-      message("ANTLR output: ${CMAKE_CURRENT_BINARY_DIR}")
       set(ANTLR_${Name}_OUTPUT_DIR
           ${CMAKE_CURRENT_BINARY_DIR}/antlr4cpp_generated_src/${ANTLR_INPUT})
     endif()

velox/connectors/clp/ClpConfig.h

@@ -27,25 +27,6 @@ class ConfigBase;
 namespace facebook::velox::connector::clp {
 class ClpConfig {
  public:
-  enum class SplitSource {
-    kLocal,
-    kS3,
-  };
-
-  static constexpr const char* kSplitSource = "clp.split-source";
-
-  [[nodiscard]] SplitSource splitSource() const {
-    auto const value = config_->get<std::string>(kSplitSource, "local");
-    auto const upperValue = boost::algorithm::to_upper_copy(value);
-    if (upperValue == "LOCAL") {
-      return SplitSource::kLocal;
-    }
-    if (upperValue == "S3") {
-      return SplitSource::kS3;
-    }
-    VELOX_UNSUPPORTED("Unsupported split source: {}.", value);
-  }
-
   explicit ClpConfig(std::shared_ptr<const config::ConfigBase> config) {
     VELOX_CHECK_NOT_NULL(config, "Config is null for CLP initialization");
     config_ = std::move(config);

velox/connectors/clp/ClpConnector.h

@@ -67,8 +67,8 @@ class ClpConnectorFactory : public ConnectorFactory {
   std::shared_ptr<Connector> newConnector(
       const std::string& id,
       std::shared_ptr<const config::ConfigBase> config,
-      folly::Executor* ioExecutor,
-      folly::Executor* cpuExecutor) override {
+      folly::Executor* /*ioExecutor*/,
+      folly::Executor* /*cpuExecutor*/) override {
     return std::make_shared<ClpConnector>(id, config);
   }
 };

velox/connectors/clp/ClpConnectorSplit.h

@@ -20,15 +20,13 @@
 
 namespace facebook::velox::connector::clp {
 struct ClpConnectorSplit : public connector::ConnectorSplit {
-  ClpConnectorSplit(
-      const std::string& connectorId,
-      const std::string& splitPath)
-      : connector::ConnectorSplit(connectorId), splitPath_(splitPath) {}
+  ClpConnectorSplit(const std::string& connectorId, const std::string& path)
+      : connector::ConnectorSplit(connectorId), path_(path) {}
 
   [[nodiscard]] std::string toString() const override {
-    return fmt::format("CLP: {}", splitPath_);
+    return fmt::format("CLP Split: {}", path_);
   }
 
-  const std::string splitPath_;
+  const std::string path_;
 };
 } // namespace facebook::velox::connector::clp

velox/connectors/clp/ClpDataSource.cpp

@@ -14,7 +14,6 @@
  * limitations under the License.
  */
 
-#include <iostream>
 #include <optional>
 
 #include "velox/connectors/clp/ClpColumnHandle.h"
@@ -35,10 +34,9 @@ ClpDataSource::ClpDataSource(
         std::shared_ptr<connector::ColumnHandle>>& columnHandles,
     velox::memory::MemoryPool* pool,
     std::shared_ptr<const ClpConfig>& clpConfig)
-    : splitSource_(clpConfig->splitSource()),
-      pool_(pool),
-      outputType_(outputType) {
+    : pool_(pool), outputType_(outputType) {
   auto clpTableHandle = std::dynamic_pointer_cast<ClpTableHandle>(tableHandle);
+  storageType_ = clpTableHandle->storageType();
   if (auto query = clpTableHandle->kqlQuery(); query && !query->empty()) {
     kqlQuery_ = *query;
   } else {
@@ -104,12 +102,12 @@ void ClpDataSource::addFieldsRecursively(
 void ClpDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
   auto clpSplit = std::dynamic_pointer_cast<ClpConnectorSplit>(split);
 
-  if (splitSource_ == ClpConfig::SplitSource::kLocal) {
+  if (storageType_ == ClpTableHandle::StorageType::kFS) {
     cursor_ = std::make_unique<search_lib::ClpCursor>(
-        clp_s::InputSource::Filesystem, clpSplit->splitPath_);
-  } else if (splitSource_ == ClpConfig::SplitSource::kS3) {
+        clp_s::InputSource::Filesystem, clpSplit->path_);
+  } else if (storageType_ == ClpTableHandle::StorageType::kS3) {
     cursor_ = std::make_unique<search_lib::ClpCursor>(
-        clp_s::InputSource::Network, clpSplit->splitPath_);
+        clp_s::InputSource::Network, clpSplit->path_);
   }
 
   cursor_->executeQuery(kqlQuery_, fields_);
@@ -119,7 +117,7 @@ VectorPtr ClpDataSource::createVector(
     const TypePtr& type,
     size_t size,
     const std::vector<clp_s::BaseColumnReader*>& projectedColumns,
-    const std::shared_ptr<std::vector<size_t>>& filteredRows,
+    const std::shared_ptr<std::vector<uint64_t>>& filteredRows,
     size_t& readerIndex) {
   if (type->kind() == TypeKind::ROW) {
     std::vector<VectorPtr> children;
@@ -155,8 +153,7 @@ VectorPtr ClpDataSource::createVector(
 std::optional<RowVectorPtr> ClpDataSource::next(
     uint64_t size,
     ContinueFuture& future) {
-  std::shared_ptr<std::vector<uint64_t>> filteredRows =
-      std::make_shared<std::vector<uint64_t>>();
+  auto filteredRows = std::make_shared<std::vector<uint64_t>>();
   auto rowsScanned = cursor_->fetchNext(size, filteredRows);
   auto rowsFiltered = filteredRows->size();
   if (rowsFiltered == 0) {

velox/connectors/clp/ClpDataSource.h

@@ -20,6 +20,7 @@
 
 #include "velox/connectors/Connector.h"
 #include "velox/connectors/clp/ClpConfig.h"
+#include "velox/connectors/clp/ClpTableHandle.h"
 #include "velox/connectors/clp/search_lib/ClpCursor.h"
 
 namespace facebook::velox::connector::clp {
@@ -87,10 +88,10 @@ class ClpDataSource : public DataSource {
       const TypePtr& type,
       size_t size,
       const std::vector<clp_s::BaseColumnReader*>& projectedColumns,
-      const std::shared_ptr<std::vector<size_t>>& filteredRows,
+      const std::shared_ptr<std::vector<uint64_t>>& filteredRows,
       size_t& readerIndex);
 
-  ClpConfig::SplitSource splitSource_;
+  ClpTableHandle::StorageType storageType_;
   std::string kqlQuery_;
   velox::memory::MemoryPool* pool_;
   RowTypePtr outputType_;

velox/connectors/clp/ClpTableHandle.h

@@ -21,18 +21,29 @@
 namespace facebook::velox::connector::clp {
 class ClpTableHandle : public ConnectorTableHandle {
  public:
+  enum class StorageType {
+    kFS,
+    kS3,
+  };
+
   ClpTableHandle(
-      std::string connectorId,
+      const std::string& connectorId,
       const std::string& tableName,
+      StorageType storageType,
       std::shared_ptr<std::string> kqlQuery)
       : ConnectorTableHandle(connectorId),
         tableName_(tableName),
+        storageType_(storageType),
         kqlQuery_(std::move(kqlQuery)) {}
 
   [[nodiscard]] const std::string& tableName() const {
     return tableName_;
   }
 
+  [[nodiscard]] const StorageType storageType() const {
+    return storageType_;
+  }
+
   [[nodiscard]] const std::shared_ptr<std::string>& kqlQuery() const {
     return kqlQuery_;
   }
@@ -43,6 +54,7 @@ class ClpTableHandle : public ConnectorTableHandle {
 
  private:
   const std::string tableName_;
+  const StorageType storageType_;
   std::shared_ptr<std::string> kqlQuery_;
 };
 } // namespace facebook::velox::connector::clp

velox/connectors/clp/search_lib/ClpCursor.cpp

@@ -101,12 +101,13 @@ uint64_t ClpCursor::fetchNext(
   return 0;
 }
 
-std::vector<clp_s::BaseColumnReader*>& ClpCursor::getProjectedColumns() const {
+const std::vector<clp_s::BaseColumnReader*>& ClpCursor::getProjectedColumns()
+    const {
   if (queryRunner_) {
     return queryRunner_->getProjectedColumns();
   }
-  static std::vector<clp_s::BaseColumnReader*> empty;
-  return empty;
+  static std::vector<clp_s::BaseColumnReader*> kEmpty;
+  return kEmpty;
 }
 
 ErrorCode ClpCursor::preprocessQuery() {

velox/connectors/clp/search_lib/ClpCursor.h

@@ -37,7 +37,7 @@ enum class ErrorCode {
   InternalError
 };
 
-enum class ColumnType { String, Integer, Float, Array, Boolean, Unknown };
+enum class ColumnType { String, Integer, Float, Array, Boolean, Unknown = -1 };
 
 struct Field {
   ColumnType type;
@@ -87,7 +87,7 @@ class ClpCursor {
    * @return A vector of BaseColumnReader pointers representing the projected
    * columns.
    */
-  std::vector<clp_s::BaseColumnReader*>& getProjectedColumns() const;
+  const std::vector<clp_s::BaseColumnReader*>& getProjectedColumns() const;
 
  private:
   /**

velox/connectors/clp/search_lib/ClpQueryRunner.cpp

@@ -61,7 +61,7 @@ void ClpQueryRunner::init(
 
 uint64_t ClpQueryRunner::fetchNext(
     size_t numRows,
-    const std::shared_ptr<std::vector<size_t>>& filteredRowIndices) {
+    const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices) {
   size_t rowsfiltered = 0;
   size_t rowsScanned = 0;
   while (curMessage_ < numMessages_) {

velox/connectors/clp/tests/ClpConnectorTest.cpp

@@ -60,9 +60,7 @@ class ClpConnectorTest : public exec::test::OperatorTestBase {
     OperatorTestBase::TearDown();
   }
 
-  exec::Split makeClpSplit(
-      const std::string& tableName,
-      const std::string& splitPath) {
+  exec::Split makeClpSplit(const std::string& splitPath) {
     return exec::Split(
         std::make_shared<ClpConnectorSplit>(kClpConnectorId, splitPath));
   }
@@ -76,7 +74,7 @@ class ClpConnectorTest : public exec::test::OperatorTestBase {
   }
 
   static std::string getExampleFilePath(const std::string& filePath) {
-    std::string current_path = fs::current_path().c_str();
+    std::string current_path = fs::current_path().string();
     return current_path + "/examples/" + filePath;
   }
 };
@@ -88,7 +86,10 @@ TEST_F(ClpConnectorTest, test1NoPushdown) {
                       ROW({"requestId", "userId", "method"},
                           {VARCHAR(), VARCHAR(), VARCHAR()}))
                   .tableHandle(std::make_shared<ClpTableHandle>(
-                      kClpConnectorId, "test_1", nullptr))
+                      kClpConnectorId,
+                      "test_1",
+                      ClpTableHandle::StorageType::kFS,
+                      nullptr))
                   .assignments({
                       {"requestId",
                        std::make_shared<ClpColumnHandle>(
@@ -104,8 +105,8 @@ TEST_F(ClpConnectorTest, test1NoPushdown) {
                   .filter("method = 'GET'")
                   .planNode();
 
-  auto output = getResults(
-      plan, {makeClpSplit("test_1", getExampleFilePath("test_1.clps"))});
+  auto output =
+      getResults(plan, {makeClpSplit(getExampleFilePath("test_1.clps"))});
   auto expected = makeRowVector(
       {// requestId
        makeFlatVector<StringView>(
@@ -133,6 +134,7 @@ TEST_F(ClpConnectorTest, test1Pushdown) {
                   .tableHandle(std::make_shared<ClpTableHandle>(
                       kClpConnectorId,
                       "test_1",
+                      ClpTableHandle::StorageType::kFS,
                       std::make_shared<std::string>(
                           "method: \"POST\" AND status: 200")))
                   .assignments({
@@ -149,8 +151,8 @@ TEST_F(ClpConnectorTest, test1Pushdown) {
                   .endTableScan()
                   .planNode();
 
-  auto output = getResults(
-      plan, {makeClpSplit("test_1", getExampleFilePath("test_1.clps"))});
+  auto output =
+      getResults(plan, {makeClpSplit(getExampleFilePath("test_1.clps"))});
   auto expected =
       makeRowVector({// requestId
                      makeFlatVector<StringView>({"req-106"}),
@@ -171,7 +173,10 @@ TEST_F(ClpConnectorTest, test2NoPushdown) {
                    ROW({"type", "subtype", "severity"},
                        {VARCHAR(), VARCHAR(), VARCHAR()})}))
           .tableHandle(std::make_shared<ClpTableHandle>(
-              kClpConnectorId, "test_2", nullptr))
+              kClpConnectorId,
+              "test_2",
+              ClpTableHandle::StorageType::kFS,
+              nullptr))
           .assignments(
               {{"timestamp",
                 std::make_shared<ClpColumnHandle>(
@@ -190,8 +195,8 @@ TEST_F(ClpConnectorTest, test2NoPushdown) {
               "(event.type = 'storage' AND event.subtype LIKE 'disk_usage%'))")
           .planNode();
 
-  auto output = getResults(
-      plan, {makeClpSplit("test_2", getExampleFilePath("test_2.clps"))});
+  auto output =
+      getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))});
   auto expected =
       makeRowVector({// timestamp
                      makeFlatVector<StringView>({"2025-04-30T08:50:05Z"}),
@@ -219,6 +224,7 @@ TEST_F(ClpConnectorTest, test2Pushdown) {
           .tableHandle(std::make_shared<ClpTableHandle>(
               kClpConnectorId,
               "test_2",
+              ClpTableHandle::StorageType::kFS,
               std::make_shared<std::string>(
                   "(event.severity: \"WARNING\" OR event.severity: \"ERROR\") AND "
                   "((event.type: \"network\" AND event.subtype: \"connection\") OR "
@@ -237,8 +243,8 @@ TEST_F(ClpConnectorTest, test2Pushdown) {
           .endTableScan()
           .planNode();
 
-  auto output = getResults(
-      plan, {makeClpSplit("test_2", getExampleFilePath("test_2.clps"))});
+  auto output =
+      getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))});
   auto expected =
       makeRowVector({// timestamp
                      makeFlatVector<StringView>({"2025-04-30T08:50:05Z"}),
@@ -266,6 +272,7 @@ TEST_F(ClpConnectorTest, test2Hybrid) {
           .tableHandle(std::make_shared<ClpTableHandle>(
               kClpConnectorId,
               "test_2",
+              ClpTableHandle::StorageType::kFS,
               std::make_shared<std::string>(
                   "((event.type: \"network\" AND event.subtype: \"connection\") OR "
                   "(event.type: \"storage\" AND event.subtype: \"disk*\"))")))
@@ -284,8 +291,8 @@ TEST_F(ClpConnectorTest, test2Hybrid) {
           .filter("upper(event.severity) IN ('WARNING', 'ERROR')")
           .planNode();
 
-  auto output = getResults(
-      plan, {makeClpSplit("test_2", getExampleFilePath("test_2.clps"))});
+  auto output =
+      getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))});
   auto expected =
       makeRowVector({// timestamp
                      makeFlatVector<StringView>({"2025-04-30T08:50:05Z"}),