refactor: Extract BaseClpCursor interface to support upcoming IR integration.

velox/connectors/clp/ClpDataSource.cpp

@@ -19,11 +19,10 @@
 #include "velox/connectors/clp/ClpColumnHandle.h"
 #include "velox/connectors/clp/ClpConnectorSplit.h"
 #include "velox/connectors/clp/ClpDataSource.h"
-
-#include "search_lib/ClpS3AuthProviderBase.h"
 #include "velox/connectors/clp/ClpTableHandle.h"
-#include "velox/connectors/clp/search_lib/ClpCursor.h"
+#include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h"
 #include "velox/connectors/clp/search_lib/ClpVectorLoader.h"
+#include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h"
 #include "velox/vector/FlatVector.h"
 
 namespace facebook::velox::connector::clp {
@@ -104,13 +103,23 @@ void ClpDataSource::addFieldsRecursively(
 void ClpDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
   auto clpSplit = std::dynamic_pointer_cast<ClpConnectorSplit>(split);
 
-  if (storageType_ == ClpConfig::StorageType::kFs) {
-    cursor_ = std::make_unique<search_lib::ClpCursor>(
-        clp_s::InputSource::Filesystem, clpSplit->path_);
-  } else if (storageType_ == ClpConfig::StorageType::kS3) {
-    cursor_ = std::make_unique<search_lib::ClpCursor>(
-        clp_s::InputSource::Network,
-        s3AuthProvider_->constructS3Url(clpSplit->path_));
+  std::string splitPath = clpSplit->path_;
+  clp_s::InputSource inputSource;
+  if (ClpConfig::StorageType::kFs == storageType_) {
+    inputSource = clp_s::InputSource::Filesystem;
+  } else if (ClpConfig::StorageType::kS3 == storageType_) {
+    inputSource = clp_s::InputSource::Network;
+    splitPath = s3AuthProvider_->constructS3Url(clpSplit->path_);
+  } else {
+    VELOX_UNREACHABLE();
+  }
+
+  if (ClpConnectorSplit::SplitType::kArchive == clpSplit->type_) {
+    cursor_ =
+        std::make_unique<search_lib::ClpArchiveCursor>(inputSource, splitPath);
+  } else {
+    VELOX_UNSUPPORTED(
+        "Unsupported split type: {}", static_cast<int>(clpSplit->type_));
   }
 
   auto pushDownQuery = clpSplit->kqlQuery_;
@@ -154,7 +163,10 @@ VectorPtr ClpDataSource::createVector(
       vectorType,
       vectorSize,
       std::make_unique<search_lib::ClpVectorLoader>(
-          projectedColumn, projectedType, filteredRows),
+          projectedColumn,
+          projectedType,
+          filteredRows,
+          cursor_->getSplitType()),
       std::move(vector));
 }
 

velox/connectors/clp/ClpDataSource.h

@@ -20,7 +20,7 @@
 
 #include "velox/connectors/Connector.h"
 #include "velox/connectors/clp/ClpConfig.h"
-#include "velox/connectors/clp/search_lib/ClpCursor.h"
+#include "velox/connectors/clp/search_lib/BaseClpCursor.h"
 
 namespace clp_s {
 class BaseColumnReader;
@@ -102,7 +102,7 @@ class ClpDataSource : public DataSource {
 
   std::vector<search_lib::Field> fields_;
 
-  std::unique_ptr<search_lib::ClpCursor> cursor_;
+  std::unique_ptr<search_lib::BaseClpCursor> cursor_;
   std::shared_ptr<ClpS3AuthProviderBase> s3AuthProvider_;
 };
 

velox/connectors/clp/search_lib/BaseClpCursor.cpp

@@ -0,0 +1,78 @@
+/*
+ * 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 <sstream>
+
+#include "velox/connectors/clp/search_lib/BaseClpCursor.h"
+
+#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/kql/kql.hpp"
+
+using namespace clp_s;
+using namespace clp_s::search;
+using namespace clp_s::search::ast;
+
+namespace facebook::velox::connector::clp::search_lib {
+
+void BaseClpCursor::executeQuery(
+    const std::string& query,
+    const std::vector<Field>& outputColumns) {
+  query_ = query;
+  outputColumns_ = outputColumns;
+  errorCode_ = preprocessQuery();
+}
+
+ErrorCode BaseClpCursor::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<EmptyExpr>(expr_)) {
+    VLOG(2) << "Query '" << query_ << "' is logically false";
+    return ErrorCode::LogicalError;
+  }
+
+  OrOfAndForm standardizePass;
+  if (expr_ = standardizePass.run(expr_);
+      std::dynamic_pointer_cast<EmptyExpr>(expr_)) {
+    VLOG(2) << "Query '" << query_ << "' is logically false";
+    return ErrorCode::LogicalError;
+  }
+
+  NarrowTypes narrowPass;
+  if (expr_ = narrowPass.run(expr_);
+      std::dynamic_pointer_cast<EmptyExpr>(expr_)) {
+    VLOG(2) << "Query '" << query_ << "' is logically false";
+    return ErrorCode::LogicalError;
+  }
+
+  ConvertToExists convertPass;
+  if (expr_ = convertPass.run(expr_);
+      std::dynamic_pointer_cast<EmptyExpr>(expr_)) {
+    VLOG(2) << "Query '" << query_ << "' is logically false";
+    return ErrorCode::LogicalError;
+  }
+
+  return ErrorCode::Success;
+}
+
+} // namespace facebook::velox::connector::clp::search_lib

velox/connectors/clp/search_lib/ClpCursor.h → velox/connectors/clp/search_lib/BaseClpCursor.h

@@ -16,22 +16,18 @@
 
 #pragma once
 
+#include <memory>
 #include <string>
+#include <string_view>
 #include <vector>
 
-#include "velox/connectors/clp/search_lib/ClpQueryRunner.h"
+#include "clp_s/InputConfig.hpp"
+#include "velox/connectors/clp/ClpConnectorSplit.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
@@ -65,14 +61,20 @@ struct Field {
 };
 
 /// 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
+/// split (archive or IR), including parsing and validating the query, loading
+/// the relevant splits, applying filters, and iterating over the results. It
+/// abstracts away the low-level details of split access
 /// while supporting projection and batch-oriented retrieval of filtered rows.
-class ClpCursor {
+class BaseClpCursor {
  public:
-  explicit ClpCursor(clp_s::InputSource inputSource, std::string archivePath);
-  ~ClpCursor();
+  explicit BaseClpCursor(
+      clp_s::InputSource inputSource,
+      std::string_view splitPath)
+      : errorCode_(ErrorCode::QueryNotInitialized),
+        inputSource_(inputSource),
+        splitPath_(std::string(splitPath)),
+        splitType_(ClpConnectorSplit::SplitType::kArchive) {}
+  virtual ~BaseClpCursor() = default;
 
   /// Executes a query. This function parses, validates, and prepares the given
   /// query for execution.
@@ -84,50 +86,52 @@ class ClpCursor {
       const std::string& query,
       const std::vector<Field>& 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.
+  /// Fetches the next set of rows from the cursor. If the split is 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(
+  virtual uint64_t fetchNext(
       uint64_t numRows,
-      const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices);
+      const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices) = 0;
 
   /// Retrieves the projected columns.
   ///
   /// @return A vector of BaseColumnReader pointers representing the projected
   /// columns.
-  const std::vector<clp_s::BaseColumnReader*>& getProjectedColumns() const;
+  virtual const std::vector<clp_s::BaseColumnReader*>& getProjectedColumns()
+      const = 0;
 
- private:
-  /// Preprocesses the query, performing parsing, validation, and optimization.
+  /// Get the type of the split that the cursor is processing.
   ///
-  /// @return The error code.
-  ErrorCode preprocessQuery();
+  /// @return The split type.
+  ClpConnectorSplit::SplitType getSplitType() const {
+    return splitType_;
+  }
 
-  /// Loads the archive at the current index.
+ protected:
   ///
   /// @return The error code.
-  ErrorCode loadArchive();
+  virtual ErrorCode loadSplit() = 0;
 
   ErrorCode errorCode_;
 
   clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem};
-  std::string archivePath_;
+  std::string splitPath_;
+  ClpConnectorSplit::SplitType splitType_;
   std::string query_;
   std::vector<Field> outputColumns_;
-  std::vector<int32_t> matchedSchemas_;
-  size_t currentSchemaIndex_{0};
-  int32_t currentSchemaId_{-1};
-  bool currentSchemaTableLoaded_{false};
-  bool currentArchiveLoaded_{false};
+
+  bool currentSplitLoaded_{false};
 
   std::shared_ptr<clp_s::search::ast::Expression> expr_;
-  std::shared_ptr<clp_s::search::SchemaMatch> schemaMatch_;
-  std::shared_ptr<ClpQueryRunner> queryRunner_;
-  std::shared_ptr<clp_s::search::Projection> projection_;
-  std::shared_ptr<clp_s::ArchiveReader> archiveReader_;
+
+ private:
+  /// Preprocesses the query, performing parsing, validation, and optimization.
+  ///
+  /// @return The error code.
+  ErrorCode preprocessQuery();
 };
 
 } // namespace facebook::velox::connector::clp::search_lib

velox/connectors/clp/search_lib/CMakeLists.txt

@@ -14,24 +14,17 @@
 velox_add_library(
   clp-s-search
   STATIC
-  ClpCursor.cpp
-  ClpCursor.h
+  BaseClpCursor.cpp
+  BaseClpCursor.h
   ClpPackageS3AuthProvider.cpp
   ClpPackageS3AuthProvider.h
-  ClpQueryRunner.cpp
-  ClpQueryRunner.h
   ClpS3AuthProviderBase.cpp
   ClpS3AuthProviderBase.h
   ClpVectorLoader.cpp
   ClpVectorLoader.h)
 
-velox_link_libraries(
-  clp-s-search
-  PUBLIC clp_s::archive_reader
-  PRIVATE
-    clp_s::clp_dependencies
-    clp_s::io
-    clp_s::search
-    clp_s::search::kql
-    velox_vector)
+add_subdirectory(archive)
+
+velox_link_libraries(clp-s-search PUBLIC clp-s-archive-search)
+
 target_compile_features(clp-s-search PRIVATE cxx_std_20)

velox/connectors/clp/search_lib/ClpVectorLoader.cpp

@@ -21,6 +21,7 @@
 #include "clp_s/ColumnReader.hpp"
 #include "clp_s/SchemaTree.hpp"
 
+#include "velox/connectors/clp/search_lib/BaseClpCursor.h"
 #include "velox/connectors/clp/search_lib/ClpVectorLoader.h"
 #include "velox/type/Timestamp.h"
 #include "velox/vector/ComplexVector.h"
@@ -124,10 +125,12 @@ auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp {
 ClpVectorLoader::ClpVectorLoader(
     clp_s::BaseColumnReader* columnReader,
     ColumnType nodeType,
-    std::shared_ptr<std::vector<uint64_t>> filteredRowIndices)
+    std::shared_ptr<std::vector<uint64_t>> filteredRowIndices,
+    ClpConnectorSplit::SplitType splitType)
     : columnReader_(columnReader),
       nodeType_(nodeType),
-      filteredRowIndices_(std::move(filteredRowIndices)) {}
+      filteredRowIndices_(std::move(filteredRowIndices)),
+      splitType_(splitType) {}
 
 template <typename T, typename VectorPtr>
 void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) {

velox/connectors/clp/search_lib/ClpVectorLoader.h

@@ -16,10 +16,12 @@
 
 #pragma once
 
+#include <simdjson.h>
+
 #include "clp_s/ColumnReader.hpp"
 #include "clp_s/SchemaTree.hpp"
+#include "velox/connectors/clp/ClpConnectorSplit.h"
 
-#include "velox/connectors/clp/search_lib/ClpCursor.h"
 #include "velox/type/Timestamp.h"
 #include "velox/vector/FlatVector.h"
 #include "velox/vector/LazyVector.h"
@@ -30,6 +32,8 @@ class BaseColumnReader;
 
 namespace facebook::velox::connector::clp::search_lib {
 
+enum class ColumnType;
+
 /// 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.
@@ -38,7 +42,8 @@ class ClpVectorLoader : public VectorLoader {
   ClpVectorLoader(
       clp_s::BaseColumnReader* columnReader,
       ColumnType nodeType,
-      std::shared_ptr<std::vector<uint64_t>> filteredRowIndices);
+      std::shared_ptr<std::vector<uint64_t>> filteredRowIndices,
+      ClpConnectorSplit::SplitType splitType);
 
  private:
   void loadInternal(
@@ -58,6 +63,7 @@ class ClpVectorLoader : public VectorLoader {
   clp_s::BaseColumnReader* columnReader_;
   ColumnType nodeType_;
   std::shared_ptr<std::vector<uint64_t>> filteredRowIndices_;
+  ClpConnectorSplit::SplitType splitType_;
 
   inline static thread_local std::unique_ptr<simdjson::ondemand::parser>
       arrayParser_ = std::make_unique<simdjson::ondemand::parser>();