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

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

@@ -26,7 +26,7 @@ struct ClpConnectorSplit : public connector::ConnectorSplit {
       : connector::ConnectorSplit(connectorId), splitPath_(splitPath) {}
 
   [[nodiscard]] std::string toString() const override {
-    return fmt::format("CLP: {}", splitPath_);
+    return fmt::format("CLP Split: {}", splitPath_);
   }
 
   const std::string splitPath_;

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,10 +102,10 @@ 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) {
+  } else if (storageType_ == ClpTableHandle::StorageType::kS3) {
     cursor_ = std::make_unique<search_lib::ClpCursor>(
         clp_s::InputSource::Network, clpSplit->splitPath_);
   }
@@ -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:
   /**