refactor: Extract BaseClpCursor interface to support upcoming IR integration.

velox/connectors/clp/ClpDataSource.cpp

@@ -19,11 +19,9 @@
 #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/ClpVectorLoader.h"
+#include "velox/connectors/clp/search_lib/ClpS3AuthProviderBase.h"
+#include "velox/connectors/clp/search_lib/archive/ClpArchiveCursor.h"
 #include "velox/vector/FlatVector.h"
 
 namespace facebook::velox::connector::clp {
@@ -104,13 +102,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_;
@@ -121,63 +129,17 @@ void ClpDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
   }
 }
 
-VectorPtr ClpDataSource::createVector(
-    const TypePtr& vectorType,
-    size_t vectorSize,
-    const std::vector<clp_s::BaseColumnReader*>& projectedColumns,
-    const std::shared_ptr<std::vector<uint64_t>>& filteredRows,
-    size_t& readerIndex) {
-  if (vectorType->kind() == TypeKind::ROW) {
-    std::vector<VectorPtr> children;
-    auto& rowType = vectorType->as<TypeKind::ROW>();
-    for (uint32_t i = 0; i < rowType.size(); ++i) {
-      children.push_back(createVector(
-          rowType.childAt(i),
-          vectorSize,
-          projectedColumns,
-          filteredRows,
-          readerIndex));
-    }
-    return std::make_shared<RowVector>(
-        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<LazyVector>(
-      pool_,
-      vectorType,
-      vectorSize,
-      std::make_unique<search_lib::ClpVectorLoader>(
-          projectedColumn, projectedType, filteredRows),
-      std::move(vector));
-}
-
 std::optional<RowVectorPtr> ClpDataSource::next(
     uint64_t size,
     ContinueFuture& future) {
-  auto filteredRows = std::make_shared<std::vector<uint64_t>>();
-  auto rowsScanned = cursor_->fetchNext(size, filteredRows);
-  auto rowsFiltered = filteredRows->size();
+  auto rowsScanned = cursor_->fetchNext(size);
+  auto rowsFiltered = cursor_->getNumFilteredRows();
   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<RowVector>(createVector(
-      outputType_, rowsFiltered, projectedColumns, filteredRows, readerIndex));
+  return std::dynamic_pointer_cast<RowVector>(
+      cursor_->createVector(pool_, outputType_, rowsFiltered));
 }
 
 } // namespace facebook::velox::connector::clp

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;
@@ -74,25 +74,6 @@ class ClpDataSource : public DataSource {
       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<clp_s::BaseColumnReader*>& projectedColumns,
-      const std::shared_ptr<std::vector<uint64_t>>& filteredRows,
-      size_t& readerIndex);
-
   ClpConfig::StorageType storageType_;
   velox::memory::MemoryPool* pool_;
   RowTypePtr outputType_;
@@ -102,7 +83,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 <glog/logging.h>
+#include <sstream>
+
+#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"
+#include "velox/connectors/clp/search_lib/BaseClpCursor.h"
+
+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,19 @@ 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)) {}
+  virtual ~BaseClpCursor() = default;
 
   /// Executes a query. This function parses, validates, and prepares the given
   /// query for execution.
@@ -84,50 +85,53 @@ 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(
-      uint64_t numRows,
-      const std::shared_ptr<std::vector<uint64_t>>& filteredRowIndices);
+  virtual uint64_t fetchNext(uint64_t numRows) = 0;
 
-  /// Retrieves the projected columns.
+  /// Gets the count of rows that satisfy the query (used to size the result
+  /// vector).
   ///
-  /// @return A vector of BaseColumnReader pointers representing the projected
-  /// columns.
-  const std::vector<clp_s::BaseColumnReader*>& getProjectedColumns() const;
+  /// @return Count of rows matching the query.
+  virtual size_t getNumFilteredRows() const = 0;
 
- private:
-  /// Preprocesses the query, performing parsing, validation, and optimization.
+  /// Creates a Vector of the specified type and size.
   ///
-  /// @return The error code.
-  ErrorCode preprocessQuery();
-
-  /// Loads the archive at the current index.
+  /// 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 pool The memory pool used by ClpDataSource to create the vector
+  /// @param vectorType
+  /// @param vectorSize
+  /// @return A Vector of the specified type and size.
+  virtual VectorPtr createVector(
+      memory::MemoryPool* pool,
+      const TypePtr& vectorType,
+      size_t vectorSize) = 0;
+
+ protected:
+  /// Loads the split from archive or IR stream.
   ///
   /// @return The error code.
-  ErrorCode loadArchive();
+  virtual ErrorCode loadSplit() = 0;
 
+  bool currentSplitLoaded_{false};
   ErrorCode errorCode_;
-
-  clp_s::InputSource inputSource_{clp_s::InputSource::Filesystem};
-  std::string archivePath_;
-  std::string query_;
+  std::shared_ptr<clp_s::search::ast::Expression> expr_;
+  clp_s::InputSource inputSource_;
   std::vector<Field> outputColumns_;
-  std::vector<int32_t> matchedSchemas_;
-  size_t currentSchemaIndex_{0};
-  int32_t currentSchemaId_{-1};
-  bool currentSchemaTableLoaded_{false};
-  bool currentArchiveLoaded_{false};
+  std::string query_;
+  std::string splitPath_;
 
-  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