diff --git a/bolt/connectors/paimon/BoltMemoryPool.h b/bolt/connectors/paimon/BoltMemoryPool.h index 8a2e1901e..7e4aac13d 100644 --- a/bolt/connectors/paimon/BoltMemoryPool.h +++ b/bolt/connectors/paimon/BoltMemoryPool.h @@ -18,18 +18,25 @@ #include #include "bolt/common/memory/MemoryPool.h" +#include "bolt/core/ExpressionEvaluator.h" namespace bytedance::bolt::connector::paimon { class BoltPaimonMemoryPool : public ::paimon::MemoryPool { public: - explicit BoltPaimonMemoryPool(memory::MemoryPool* const pool) - : pool_(std::move(pool)) {} + explicit BoltPaimonMemoryPool( + memory::MemoryPool* const pool, + core::ExpressionEvaluator* const expressionEvaluator = nullptr) + : pool_(pool), expressionEvaluator_(expressionEvaluator) {} memory::MemoryPool* getBoltPool() const { return pool_; } + core::ExpressionEvaluator* getExpressionEvaluator() const { + return expressionEvaluator_; + } + void* Malloc(uint64_t size, uint64_t alignment) override { if (alignment == 0) { return pool_->allocate(static_cast(size)); @@ -71,6 +78,7 @@ class BoltPaimonMemoryPool : public ::paimon::MemoryPool { private: memory::MemoryPool* const pool_; + core::ExpressionEvaluator* const expressionEvaluator_; }; } // namespace bytedance::bolt::connector::paimon diff --git a/bolt/connectors/paimon/PaimonDataSource.cpp b/bolt/connectors/paimon/PaimonDataSource.cpp index 125fd75ea..f431a932c 100644 --- a/bolt/connectors/paimon/PaimonDataSource.cpp +++ b/bolt/connectors/paimon/PaimonDataSource.cpp @@ -18,17 +18,22 @@ #include #include #include +#include #include #include #include #include #include #include +#include +#include #include "bolt/common/base/Exceptions.h" #include "bolt/connectors/hive/TableHandle.h" #include "bolt/connectors/paimon/BoltMemoryPool.h" #include "bolt/connectors/paimon/PaimonConfig.h" #include "bolt/connectors/paimon/PaimonFilterTranslator.h" +#include "bolt/expression/Expr.h" +#include "bolt/expression/ExprToSubfieldFilter.h" #include "bolt/type/StringView.h" #include "bolt/type/Type.h" #include "bolt/vector/BaseVector.h" @@ -38,6 +43,130 @@ #include "bolt/vector/arrow/Bridge.h" namespace bytedance::bolt::connector::paimon { +namespace { + +std::shared_ptr paimonColumnHandle( + const std::shared_ptr& columnHandle, + const std::string& outputName) { + auto handle = std::dynamic_pointer_cast(columnHandle); + BOLT_CHECK_NOT_NULL( + handle, + "ColumnHandle must be an instance of PaimonColumnHandle for {}", + outputName); + return handle; +} + +std::shared_ptr findPaimonColumnHandle( + const std::unordered_map>& + columnHandles, + const std::string& fieldName) { + auto it = columnHandles.find(fieldName); + if (it != columnHandles.end()) { + return paimonColumnHandle(it->second, fieldName); + } + + for (const auto& [outputName, columnHandle] : columnHandles) { + auto handle = paimonColumnHandle(columnHandle, outputName); + if (handle->name() == fieldName) { + return handle; + } + } + return nullptr; +} + +} // namespace + +namespace { + +core::TypedExprPtr combineConjuncts(std::vector conjuncts) { + if (conjuncts.empty()) { + return nullptr; + } + if (conjuncts.size() == 1) { + return std::move(conjuncts.front()); + } + return std::make_shared( + BOOLEAN(), std::move(conjuncts), "and"); +} + +void collectFieldNames( + const core::TypedExprPtr& expression, + std::vector& fieldNames, + std::unordered_set& seen) { + std::vector pending; + if (expression) { + pending.push_back(expression); + } + + while (!pending.empty()) { + auto current = std::move(pending.back()); + pending.pop_back(); + if (!current) { + continue; + } + + const auto* field = + dynamic_cast(current.get()); + if (field != nullptr && + (field->inputs().empty() || + dynamic_cast( + field->inputs().front().get()) != nullptr) && + seen.insert(field->name()).second) { + fieldNames.push_back(field->name()); + } + + const auto& inputs = current->inputs(); + for (size_t i = inputs.size(); i > 0; --i) { + pending.push_back(inputs[i - 1]); + } + } +} + +std::vector collectFieldNames( + const core::TypedExprPtr& expression) { + std::vector fieldNames; + std::unordered_set seen; + collectFieldNames(expression, fieldNames, seen); + return fieldNames; +} + +std::pair, core::TypedExprPtr> planFilter( + const core::TypedExprPtr& expression, + const RowTypePtr& rowType, + core::ExpressionEvaluator* evaluator) { + std::vector conjuncts; + exec::flattenTopLevelConjuncts(expression, conjuncts); + + std::vector> predicates; + std::vector remainingConjuncts; + for (const auto& conjunct : conjuncts) { + auto result = + PaimonFilterTranslator::translate(conjunct, rowType, evaluator); + if (result.ok()) { + predicates.push_back(std::move(result.value)); + } else { + VLOG(1) << "PaimonDataSource: Keeping filter conjunct " + << conjunct->toString() + << " as remaining filter: " << result.reason; + remainingConjuncts.push_back(conjunct); + } + } + + std::shared_ptr<::paimon::Predicate> predicate; + if (predicates.size() == 1) { + predicate = std::move(predicates.front()); + } else if (!predicates.empty()) { + auto result = ::paimon::PredicateBuilder::And(predicates); + BOLT_CHECK( + result.ok(), "PredicateBuilder::And failed for translated conjuncts"); + predicate = std::move(result).value(); + } + + return { + std::move(predicate), combineConjuncts(std::move(remainingConjuncts))}; +} + +} // namespace PaimonDataSource::PaimonDataSource( const std::shared_ptr& outputType, @@ -49,21 +178,47 @@ PaimonDataSource::PaimonDataSource( const std::shared_ptr& paimonConfig) : outputType_(outputType), tableHandle_(std::dynamic_pointer_cast(tableHandle)), + expressionEvaluator_(queryCtx->expressionEvaluator()), pool_(queryCtx->memoryPool()) { // Wrap the query-context pool for Paimon's native memory management. - paimonPool_ = std::make_shared(pool_); + paimonPool_ = + std::make_shared(pool_, expressionEvaluator_); ::paimon::ReadContextBuilder ctxBuilder(tableHandle_->tablePath()); std::vector columns; columns.reserve(outputType_->size()); + std::vector filterTypes; + filterTypes.reserve(outputType_->size()); + std::unordered_set readColumnNames; for (const auto& outName : outputType_->names()) { auto it = columnHandles.find(outName); BOLT_CHECK( it != columnHandles.end(), "Could not find column handle with name: {}", outName); - columns.push_back(it->second->name()); + const auto& columnName = paimonColumnHandle(it->second, outName)->name(); + columns.push_back(columnName); + readColumnNames.insert(columnName); + } + for (const auto& type : outputType_->children()) { + filterTypes.push_back(type); + } + + if (tableHandle_->filter()) { + for (const auto& fieldName : collectFieldNames(tableHandle_->filter())) { + auto columnHandle = findPaimonColumnHandle(columnHandles, fieldName); + BOLT_CHECK_NOT_NULL( + columnHandle, + "Could not find column handle for filter field: {}", + fieldName); + if (readColumnNames.insert(columnHandle->name()).second) { + columns.push_back(columnHandle->name()); + filterTypes.push_back(columnHandle->type()); + } + } } + auto filterNames = columns; + filterRowType_ = ROW(std::move(filterNames), std::move(filterTypes)); VLOG(1) << "PaimonDataSource::PaimonDataSource(): Read schema: " << folly::join(", ", columns); ctxBuilder.SetReadSchema(columns); @@ -126,44 +281,17 @@ PaimonDataSource::PaimonDataSource( ctxBuilder.SetPrefetchMaxParallelNum(paimonConfig->prefetchMaxParallel()); } - // Translate the filter expression from PaimonTableHandle into a - // paimon-native Predicate for pushdown into the parquet reader. - // - // The filter expression references columns by their *real* paimon schema - // names (e.g., "id", "name") as set by the SparkSQL/gluten planner, but - // outputType_ carries internal alias names (e.g., "n0_0", "n0_1"). - // We must build a RowType whose names match the filter's field references - // so that extractFieldInfo can resolve field indices correctly. if (tableHandle_->filter()) { - std::vector realNames; - std::vector realTypes; - realNames.reserve(outputType_->size()); - realTypes.reserve(outputType_->size()); - for (size_t i = 0; i < outputType_->size(); ++i) { - const auto& outName = outputType_->nameOf(i); - auto it = columnHandles.find(outName); - BOLT_CHECK( - it != columnHandles.end(), - "Could not find column handle with name: {}", - outName); - realNames.push_back(it->second->name()); // real paimon name - realTypes.push_back(outputType_->childAt(i)); // preserve type - } - auto filterRowType = ROW(std::move(realNames), std::move(realTypes)); - - auto result = PaimonFilterTranslator::translate( - tableHandle_->filter(), filterRowType); - if (result.ok()) { + auto filterPlan = planFilter( + tableHandle_->filter(), filterRowType_, expressionEvaluator_); + if (filterPlan.first) { VLOG(1) << "PaimonDataSource: Translated filter to paimon Predicate: " - << result.value->ToString(); - ctxBuilder.SetPredicate(result.value); - } else { - LOG(WARNING) - << "PaimonDataSource: Could not fully translate filter expression " - << tableHandle_->filter()->toString() - << " to paimon Predicate: " << result.reason - << "; filter pushdown will not be applied"; - ctxBuilder.SetPredicate(nullptr); + << filterPlan.first->ToString(); + } + ctxBuilder.SetPredicate(filterPlan.first); + if (filterPlan.second) { + remainingFilterExprSet_ = + expressionEvaluator_->compile(filterPlan.second); } } else { ctxBuilder.SetPredicate(nullptr); @@ -275,25 +403,17 @@ std::optional PaimonDataSource::next( } ArrowOptions opts; - auto vec = bytedance::bolt::importFromArrowAsOwner(sch, arr, opts, pool_); - - const auto& row = std::dynamic_pointer_cast(vec); + auto row = std::dynamic_pointer_cast( + bytedance::bolt::importFromArrowAsOwner(sch, arr, opts, pool_)); BOLT_CHECK(row != nullptr, "Imported vector is not a RowVector"); const auto& rowType = row->type()->asRow(); VLOG(1) << "Imported RowVector size: " << row->size() << ", number of fields: " << rowType.size(); - // If we have _VALUE_KIND as the first field, drop it - but only if the - // original Parquet reader actually exported it. Check if the data actually - // exists in the child vector before proceeding. - auto firstChild = row->childAt(0); - VLOG(1) << "First child vector type: " << firstChild->type()->toString() - << ", elements: " << firstChild->size(); if (rowType.nameOf(0) == "_VALUE_KIND" && rowType.size() > 1) { VLOG(1) << "Dropping _VALUE_KIND field"; - // Create a new row vector without the _VALUE_KIND field std::vector newChildren; for (int i = 1; i < rowType.size(); ++i) { newChildren.push_back(row->childAt(i)); @@ -306,60 +426,46 @@ std::optional PaimonDataSource::next( newTypes.push_back(rowType.childAt(i)); } - const auto& newRowType = ROW(std::move(newNames), std::move(newTypes)); - auto newRowVec = std::make_shared( - pool_, newRowType, nullptr, row->size(), newChildren); - - // Copy null information - newRowVec->setNulls(row->nulls()); - - VLOG(1) << "New RowVector size: " << newRowVec->size() - << ", number of fields: " << newRowType->size(); - // VLOG(1) << newRowVec->toPrettyString(); - completedRows_ += newRowVec->size(); - - // Re-wrap with outputType_ to ensure internal alias names match upstream - // operators' expectations (same as the non-_VALUE_KIND path below). - std::vector outputColumns; - outputColumns.reserve(outputType_->size()); - for (size_t i = 0; i < outputType_->size(); ++i) { - outputColumns.push_back(newRowVec->childAt(i)); - } - return std::make_shared( + row = std::make_shared( pool_, - outputType_, - nullptr, - newRowVec->size(), - std::move(outputColumns)); + ROW(std::move(newNames), std::move(newTypes)), + row->nulls(), + row->size(), + std::move(newChildren)); } - if (VLOG_IS_ON(1)) { - auto idColumn = row->childAt(0); - auto* idFlat = idColumn->asFlatVector(); - for (int i = 0; i < row->size(); ++i) { - if (idColumn->isNullAt(i)) { - VLOG(1) << "Row " << i << ": id = NULL"; - } else { - VLOG(1) << "Row " << i << ": id = " << idFlat->valueAt(i); - } + completedRows_ += row->size(); + auto rowsRemaining = row->size(); + BufferPtr remainingIndices; + if (remainingFilterExprSet_) { + auto filterInput = std::make_shared( + pool_, filterRowType_, row->nulls(), row->size(), row->children()); + rowsRemaining = evaluateRemainingFilter(filterInput); + if (rowsRemaining == 0) { + return RowVector::createEmpty(outputType_, pool_); + } + if (rowsRemaining < row->size()) { + remainingIndices = filterEvalCtx_.selectedIndices; } } - completedRows_ += row->size(); - VLOG(1) << "Returning row vector of size " << row->size(); - - // Wrap the imported vector in a new RowVector that uses outputType_ as its - // type. The raw Arrow import carries real column names (e.g. "id", "name") - // from the Paimon schema, but upstream operators (FilterProject, ProjectNode) - // expect internal alias names (e.g. "n0_0", "n0_1") from outputType_. Re-wrap - // by index position to preserve the correct naming contract. std::vector outputColumns; outputColumns.reserve(outputType_->size()); for (size_t i = 0; i < outputType_->size(); ++i) { - outputColumns.push_back(row->childAt(i)); + outputColumns.push_back( + exec::wrapChild(rowsRemaining, remainingIndices, row->childAt(i))); } return std::make_shared( - pool_, outputType_, nullptr, row->size(), std::move(outputColumns)); + pool_, outputType_, nullptr, rowsRemaining, std::move(outputColumns)); +} + +vector_size_t PaimonDataSource::evaluateRemainingFilter( + RowVectorPtr& rowVector) { + filterRows_.resizeFill(rowVector->size()); + expressionEvaluator_->evaluate( + remainingFilterExprSet_.get(), filterRows_, *rowVector, filterResult_); + return exec::processFilterResults( + filterResult_, filterRows_, filterEvalCtx_, pool_); } } // namespace bytedance::bolt::connector::paimon diff --git a/bolt/connectors/paimon/PaimonDataSource.h b/bolt/connectors/paimon/PaimonDataSource.h index 92c9e5f11..4d3fe1cec 100644 --- a/bolt/connectors/paimon/PaimonDataSource.h +++ b/bolt/connectors/paimon/PaimonDataSource.h @@ -20,6 +20,9 @@ #include "bolt/connectors/paimon/PaimonConfig.h" #include "bolt/connectors/paimon/PaimonConnectorSplit.h" #include "bolt/connectors/paimon/PaimonTableHandle.h" +#include "bolt/core/ExpressionEvaluator.h" +#include "bolt/exec/OperatorUtils.h" +#include "bolt/vector/SelectivityVector.h" // Forward declare paimon types namespace paimon { @@ -64,8 +67,12 @@ class PaimonDataSource : public DataSource { override {} private: + vector_size_t evaluateRemainingFilter(RowVectorPtr& rowVector); + std::shared_ptr outputType_; + RowTypePtr filterRowType_; std::shared_ptr tableHandle_; + core::ExpressionEvaluator* const expressionEvaluator_; memory::MemoryPool* pool_; std::vector> holdReader_; @@ -76,6 +83,11 @@ class PaimonDataSource : public DataSource { uint64_t completedRows_{0}; uint64_t completedBytes_{0}; + + std::unique_ptr remainingFilterExprSet_; + VectorPtr filterResult_; + SelectivityVector filterRows_; + exec::FilterEvalCtx filterEvalCtx_; }; } // namespace bytedance::bolt::connector::paimon diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.cpp b/bolt/connectors/paimon/PaimonFilterTranslator.cpp index 41464b408..5d1de0ad4 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.cpp +++ b/bolt/connectors/paimon/PaimonFilterTranslator.cpp @@ -28,6 +28,8 @@ #include #include #include "bolt/core/Expressions.h" +#include "bolt/expression/Expr.h" +#include "bolt/expression/ExprToSubfieldFilter.h" #include "bolt/type/Subfield.h" #include "bolt/type/Timestamp.h" #include "bolt/type/filter/FilterBase.h" @@ -188,6 +190,131 @@ core::TypedExprPtr makeArrayConstant( return std::make_shared(wrapped); } +VectorPtr evaluateConstantExpression( + const core::TypedExprPtr& expr, + core::ExpressionEvaluator* evaluator) { + if (!evaluator) { + return nullptr; + } + auto exprSet = evaluator->compile(expr); + if (exprSet->size() != 1 || !exprSet->exprs()[0]->isConstant()) { + return nullptr; + } + + RowVector input( + evaluator->pool(), ROW({}, {}), nullptr, 1, std::vector{}); + SelectivityVector rows(1); + VectorPtr result; + try { + evaluator->evaluate(exprSet.get(), rows, input, result); + } catch (const BoltUserError&) { + return nullptr; + } + return result; +} + +std::optional integerValueAt(const VectorPtr& vector) { + switch (vector->typeKind()) { + case TypeKind::TINYINT: + return vector->as>()->valueAt(0); + case TypeKind::SMALLINT: + return vector->as>()->valueAt(0); + case TypeKind::INTEGER: + return vector->as>()->valueAt(0); + case TypeKind::BIGINT: + return vector->as>()->valueAt(0); + default: + return std::nullopt; + } +} + +std::optional<::paimon::Literal> literalFromVector( + const VectorPtr& vector, + ::paimon::FieldType fieldType) { + if (!vector || vector->size() == 0) { + return std::nullopt; + } + if (vector->isNullAt(0)) { + return ::paimon::Literal(fieldType); + } + + switch (fieldType) { + case ::paimon::FieldType::BOOLEAN: + if (vector->typeKind() != TypeKind::BOOLEAN) { + return std::nullopt; + } + return ::paimon::Literal(vector->as>()->valueAt(0)); + case ::paimon::FieldType::TINYINT: + case ::paimon::FieldType::SMALLINT: + case ::paimon::FieldType::INT: + case ::paimon::FieldType::BIGINT: { + auto v = integerValueAt(vector); + if (!v) { + return std::nullopt; + } + if (fieldType == ::paimon::FieldType::TINYINT) { + return ::paimon::Literal(static_cast(*v)); + } + if (fieldType == ::paimon::FieldType::SMALLINT) { + return ::paimon::Literal(static_cast(*v)); + } + if (fieldType == ::paimon::FieldType::INT) { + return ::paimon::Literal(static_cast(*v)); + } + return ::paimon::Literal(*v); + } + case ::paimon::FieldType::FLOAT: { + if (vector->typeKind() == TypeKind::REAL) { + return ::paimon::Literal(vector->as>()->valueAt(0)); + } + if (vector->typeKind() == TypeKind::DOUBLE) { + return ::paimon::Literal( + static_cast(vector->as>()->valueAt(0))); + } + if (auto v = integerValueAt(vector)) { + return ::paimon::Literal(static_cast(*v)); + } + return std::nullopt; + } + case ::paimon::FieldType::DOUBLE: { + if (vector->typeKind() == TypeKind::REAL) { + return ::paimon::Literal( + static_cast(vector->as>()->valueAt(0))); + } + if (vector->typeKind() == TypeKind::DOUBLE) { + return ::paimon::Literal( + vector->as>()->valueAt(0)); + } + if (auto v = integerValueAt(vector)) { + return ::paimon::Literal(static_cast(*v)); + } + return std::nullopt; + } + case ::paimon::FieldType::STRING: + case ::paimon::FieldType::BINARY: { + if (vector->typeKind() != TypeKind::VARCHAR && + vector->typeKind() != TypeKind::VARBINARY) { + return std::nullopt; + } + auto sv = vector->as>()->valueAt(0); + return ::paimon::Literal( + ::paimon::FieldType::STRING, sv.data(), sv.size()); + } + case ::paimon::FieldType::TIMESTAMP: { + if (vector->typeKind() != TypeKind::TIMESTAMP) { + return std::nullopt; + } + auto ts = vector->as>()->valueAt(0); + int64_t totalNanos = (ts.getSeconds() * 1'000'000'000LL) + ts.getNanos(); + int64_t millis = totalNanos / 1'000'000; + auto nanoOfMillis = static_cast(totalNanos % 1'000'000); + return ::paimon::Literal(::paimon::Timestamp(millis, nanoOfMillis)); + } + default: + return std::nullopt; + } +} + } // namespace // =========================================================================== @@ -224,7 +351,8 @@ std::string PaimonFilterTranslator::normalizeOpName(const std::string& opName) { ToPaimonPredicateResult PaimonFilterTranslator::translate( const core::TypedExprPtr& expr, - const RowTypePtr& rowType) { + const RowTypePtr& rowType, + core::ExpressionEvaluator* evaluator) { auto fail = [](std::string reason) -> ToPaimonPredicateResult { LOG(WARNING) << "translate (with rowType): " << reason; return {nullptr, std::move(reason)}; @@ -239,12 +367,13 @@ ToPaimonPredicateResult PaimonFilterTranslator::translate( return fail("expr is not a CallTypedExpr"); } - return translateCall(*call, rowType); + return translateCall(*call, rowType, evaluator); } ToPaimonPredicateResult PaimonFilterTranslator::translateCall( const core::CallTypedExpr& call, - const RowTypePtr& rowType) { + const RowTypePtr& rowType, + core::ExpressionEvaluator* evaluator) { auto fail = [](std::string reason) -> ToPaimonPredicateResult { LOG(WARNING) << "translateCall (with rowType): " << reason; return {nullptr, std::move(reason)}; @@ -254,218 +383,292 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( return {std::move(pred), ""}; }; - const auto& opName = normalizeOpName(call.name()); - const auto& inputs = call.inputs(); + auto translateLeafCall = + [&](const core::CallTypedExpr& leafCall) -> ToPaimonPredicateResult { + const auto opName = normalizeOpName(leafCall.name()); + const auto& inputs = leafCall.inputs(); - // Handle AND: recursively translate all children. - if (opName == "and") { - std::vector> children; - children.reserve(inputs.size()); - for (const auto& input : inputs) { - auto child = translate(input, rowType); - if (!child.ok()) { - return fail(fmt::format("AND child failed: {}", child.reason)); - } - children.push_back(std::move(child.value)); + // Leaf predicates require at least one operand (the field). + if (inputs.empty()) { + return fail("leaf op has no inputs"); } - auto result = ::paimon::PredicateBuilder::And(children); - if (!result.ok()) { - return fail("PredicateBuilder::And failed"); + + auto fieldInfo = extractFieldInfo(inputs[0], rowType); + if (!fieldInfo) { + return fail(fmt::format( + "could not extract field info for op {}, input[0]={}", + opName, + inputs[0]->toString())); } - return ok(std::move(result).value()); - } - // Handle OR: recursively translate all children. - if (opName == "or") { - std::vector> children; - children.reserve(inputs.size()); - for (const auto& input : inputs) { - auto child = translate(input, rowType); - if (!child.ok()) { - return fail("OR child failed"); + const auto& fieldName = fieldInfo->name; + const auto fieldIndex = fieldInfo->index; + const auto& fieldType = fieldInfo->fieldType; + + // Equality: eq(field, constant) + if (opName == "eq") { + if (inputs.size() < 2) { + return fail("eq requires 2 inputs"); } - children.push_back(std::move(child.value)); - } - auto result = ::paimon::PredicateBuilder::Or(children); - if (!result.ok()) { - return fail("PredicateBuilder::Or failed"); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("eq: could not extract literal"); + } + return ok(::paimon::PredicateBuilder::Equal( + fieldIndex, fieldName, fieldType, *lit)); } - return ok(std::move(result).value()); - } - // Handle NOT: translate inner with negation applied per-operator. - if (opName == "not") { - if (inputs.empty()) { - return fail("NOT has no inputs"); - } - const auto* innerCall = - dynamic_cast(inputs[0].get()); - if (!innerCall) { - return fail("NOT inner is not a CallTypedExpr"); - } - auto negatedOp = applyNegation(normalizeOpName(innerCall->name())); - if (negatedOp.has_value()) { - core::CallTypedExpr negatedCall( - innerCall->type(), innerCall->inputs(), negatedOp.value()); - return translateCall(negatedCall, rowType); - } - if (normalizeOpName(innerCall->name()) == "is_null") { - auto fieldInfo = extractFieldInfo(innerCall->inputs()[0], rowType); - if (!fieldInfo) { - return fail("NOT is_null: could not extract field info"); + // Not equal: neq(field, constant) + if (opName == "neq") { + if (inputs.size() < 2) { + return fail("neq requires 2 inputs"); } - return ok(::paimon::PredicateBuilder::IsNotNull( - fieldInfo->index, fieldInfo->name, fieldInfo->fieldType)); - } - if (normalizeOpName(innerCall->name()) == "is_not_null") { - auto fieldInfo = extractFieldInfo(innerCall->inputs()[0], rowType); - if (!fieldInfo) { - return fail("NOT is_not_null: could not extract field info"); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("neq: could not extract literal"); } - return ok(::paimon::PredicateBuilder::IsNull( - fieldInfo->index, fieldInfo->name, fieldInfo->fieldType)); + return ok(::paimon::PredicateBuilder::NotEqual( + fieldIndex, fieldName, fieldType, *lit)); } - return fail("unsupported negation target"); - } - - // Leaf predicates require at least one operand (the field). - if (inputs.empty()) { - return fail("leaf op has no inputs"); - } - auto fieldInfo = extractFieldInfo(inputs[0], rowType); - if (!fieldInfo) { - return fail(fmt::format( - "could not extract field info for op {}, input[0]={}", - opName, - inputs[0]->toString())); - } - - const auto& fieldName = fieldInfo->name; - const auto fieldIndex = fieldInfo->index; - const auto& fieldType = fieldInfo->fieldType; - - // Equality: eq(field, constant) - if (opName == "eq") { - if (inputs.size() < 2) { - return fail("eq requires 2 inputs"); - } - auto lit = extractLiteral(inputs[1], fieldType); - if (!lit) { - return fail("eq: could not extract literal"); + // Less than: lt(field, constant) + if (opName == "lt") { + if (inputs.size() < 2) { + return fail("lt requires 2 inputs"); + } + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("lt: could not extract literal"); + } + return ok(::paimon::PredicateBuilder::LessThan( + fieldIndex, fieldName, fieldType, *lit)); } - return ok(::paimon::PredicateBuilder::Equal( - fieldIndex, fieldName, fieldType, *lit)); - } - // Not equal: neq(field, constant) - if (opName == "neq") { - if (inputs.size() < 2) { - return fail("neq requires 2 inputs"); - } - auto lit = extractLiteral(inputs[1], fieldType); - if (!lit) { - return fail("neq: could not extract literal"); + // Less or equal: lte(field, constant) + if (opName == "lte") { + if (inputs.size() < 2) { + return fail("lte requires 2 inputs"); + } + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("lte: could not extract literal"); + } + return ok(::paimon::PredicateBuilder::LessOrEqual( + fieldIndex, fieldName, fieldType, *lit)); } - return ok(::paimon::PredicateBuilder::NotEqual( - fieldIndex, fieldName, fieldType, *lit)); - } - // Less than: lt(field, constant) - if (opName == "lt") { - if (inputs.size() < 2) { - return fail("lt requires 2 inputs"); + // Greater than: gt(field, constant) + if (opName == "gt") { + if (inputs.size() < 2) { + return fail("gt requires 2 inputs"); + } + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("gt: could not extract literal"); + } + return ok(::paimon::PredicateBuilder::GreaterThan( + fieldIndex, fieldName, fieldType, *lit)); } - auto lit = extractLiteral(inputs[1], fieldType); - if (!lit) { - return fail("lt: could not extract literal"); + + // Greater or equal: gte(field, constant) + if (opName == "gte") { + if (inputs.size() < 2) { + return fail("gte requires 2 inputs"); + } + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("gte: could not extract literal"); + } + return ok(::paimon::PredicateBuilder::GreaterOrEqual( + fieldIndex, fieldName, fieldType, *lit)); } - return ok(::paimon::PredicateBuilder::LessThan( - fieldIndex, fieldName, fieldType, *lit)); - } - // Less or equal: lte(field, constant) - if (opName == "lte") { - if (inputs.size() < 2) { - return fail("lte requires 2 inputs"); + // Between: between(field, lower, upper) + if (opName == "between") { + if (inputs.size() < 3) { + return fail("between requires 3 inputs"); + } + auto low = extractLiteral(inputs[1], fieldType, evaluator); + auto high = extractLiteral(inputs[2], fieldType, evaluator); + if (!low || !high) { + return fail("between: could not extract literals"); + } + return ok(::paimon::PredicateBuilder::Between( + fieldIndex, fieldName, fieldType, *low, *high)); } - auto lit = extractLiteral(inputs[1], fieldType); - if (!lit) { - return fail("lte: could not extract literal"); + + // In: in(field, array_constant) / not_in(field, array_constant) + if (opName == "in" || opName == "not_in") { + if (inputs.size() < 2) { + return fail("in/not_in requires 2 inputs"); + } + bool negated = (opName == "not_in"); + auto result = extractInListLiterals(inputs[1], fieldType); + if (!result.has_value()) { + return fail("in/not_in: could not extract IN-list literals"); + } + auto literals = std::move(result.value()); + if (negated) { + return ok(::paimon::PredicateBuilder::NotIn( + fieldIndex, fieldName, fieldType, literals)); + } + return ok(::paimon::PredicateBuilder::In( + fieldIndex, fieldName, fieldType, literals)); } - return ok(::paimon::PredicateBuilder::LessOrEqual( - fieldIndex, fieldName, fieldType, *lit)); - } - // Greater than: gt(field, constant) - if (opName == "gt") { - if (inputs.size() < 2) { - return fail("gt requires 2 inputs"); + // Like: like(string_field, pattern). Three-arg LIKE with an explicit escape + // character is kept as a remaining filter to avoid changing escape + // semantics. + if (opName == "like") { + if (inputs.size() != 2) { + return fail("like requires 2 inputs"); + } + if (fieldType != ::paimon::FieldType::STRING && + fieldType != ::paimon::FieldType::BINARY) { + return fail("like only supports string/binary fields"); + } + auto lit = extractLiteral(inputs[1], fieldType, evaluator); + if (!lit) { + return fail("like: could not extract pattern literal"); + } + auto result = ::paimon::PredicateBuilder::Like( + fieldIndex, fieldName, fieldType, *lit); + if (!result.ok()) { + return fail("like: " + result.status().ToString()); + } + return ok(std::move(result).value()); } - auto lit = extractLiteral(inputs[1], fieldType); - if (!lit) { - return fail("gt: could not extract literal"); + + // Is null: is_null(field) + if (opName == "is_null") { + return ok( + ::paimon::PredicateBuilder::IsNull(fieldIndex, fieldName, fieldType)); } - return ok(::paimon::PredicateBuilder::GreaterThan( - fieldIndex, fieldName, fieldType, *lit)); - } - // Greater or equal: gte(field, constant) - if (opName == "gte") { - if (inputs.size() < 2) { - return fail("gte requires 2 inputs"); + // Is not null: is_not_null(field) + if (opName == "is_not_null") { + return ok(::paimon::PredicateBuilder::IsNotNull( + fieldIndex, fieldName, fieldType)); } - auto lit = extractLiteral(inputs[1], fieldType); - if (!lit) { - return fail("gte: could not extract literal"); + + return fail(fmt::format("unsupported opName={}", opName)); + }; + + auto translateNonCompoundCall = + [&](const core::CallTypedExpr& node) -> ToPaimonPredicateResult { + const auto opName = normalizeOpName(node.name()); + const auto& inputs = node.inputs(); + + // Handle NOT: translate inner with negation applied per-operator. + if (opName == "not") { + if (inputs.empty()) { + return fail("NOT has no inputs"); + } + const auto* innerCall = + dynamic_cast(inputs[0].get()); + if (!innerCall) { + return fail("NOT inner is not a CallTypedExpr"); + } + auto negatedOp = applyNegation(normalizeOpName(innerCall->name())); + if (negatedOp.has_value()) { + core::CallTypedExpr negatedCall( + innerCall->type(), innerCall->inputs(), negatedOp.value()); + return translateLeafCall(negatedCall); + } + if (normalizeOpName(innerCall->name()) == "is_null") { + if (innerCall->inputs().empty()) { + return fail("NOT is_null has no inputs"); + } + auto fieldInfo = extractFieldInfo(innerCall->inputs()[0], rowType); + if (!fieldInfo) { + return fail("NOT is_null: could not extract field info"); + } + return ok(::paimon::PredicateBuilder::IsNotNull( + fieldInfo->index, fieldInfo->name, fieldInfo->fieldType)); + } + if (normalizeOpName(innerCall->name()) == "is_not_null") { + if (innerCall->inputs().empty()) { + return fail("NOT is_not_null has no inputs"); + } + auto fieldInfo = extractFieldInfo(innerCall->inputs()[0], rowType); + if (!fieldInfo) { + return fail("NOT is_not_null: could not extract field info"); + } + return ok(::paimon::PredicateBuilder::IsNull( + fieldInfo->index, fieldInfo->name, fieldInfo->fieldType)); + } + return fail("unsupported negation target"); } - return ok(::paimon::PredicateBuilder::GreaterOrEqual( - fieldIndex, fieldName, fieldType, *lit)); - } - // Between: between(field, lower, upper) - if (opName == "between") { - if (inputs.size() < 3) { - return fail("between requires 3 inputs"); + return translateLeafCall(node); + }; + + struct Frame { + const core::CallTypedExpr* node; + bool combine; + }; + + std::vector pending{{&call, false}}; + std::vector> results; + + while (!pending.empty()) { + auto frame = pending.back(); + pending.pop_back(); + + const auto opName = normalizeOpName(frame.node->name()); + const auto& inputs = frame.node->inputs(); + const bool isCompound = opName == "and" || opName == "or"; + + if (!isCompound) { + auto leaf = translateNonCompoundCall(*frame.node); + if (!leaf.ok()) { + return leaf; + } + results.push_back(std::move(leaf.value)); + continue; } - auto low = extractLiteral(inputs[1], fieldType); - auto high = extractLiteral(inputs[2], fieldType); - if (!low || !high) { - return fail("between: could not extract literals"); + + if (!frame.combine) { + pending.push_back({frame.node, true}); + for (size_t i = inputs.size(); i > 0; --i) { + const auto& input = inputs[i - 1]; + const auto* childCall = + dynamic_cast(input.get()); + if (!childCall) { + return fail(fmt::format( + "{} child failed: expr is not a CallTypedExpr", + opName == "and" ? "AND" : "OR")); + } + pending.push_back({childCall, false}); + } + continue; } - return ok(::paimon::PredicateBuilder::Between( - fieldIndex, fieldName, fieldType, *low, *high)); - } - // In: in(field, array_constant) / not_in(field, array_constant) - if (opName == "in" || opName == "not_in") { - bool negated = (opName == "not_in"); - auto result = extractInListLiterals(inputs[1], fieldType); - if (!result.has_value()) { - return fail("in/not_in: could not extract IN-list literals"); + if (results.size() < inputs.size()) { + return fail("compound predicate result stack underflow"); } - auto literals = std::move(result.value()); - if (negated) { - return ok(::paimon::PredicateBuilder::NotIn( - fieldIndex, fieldName, fieldType, literals)); + const auto firstChild = results.size() - inputs.size(); + std::vector> children; + children.reserve(inputs.size()); + for (size_t i = firstChild; i < results.size(); ++i) { + children.push_back(std::move(results[i])); } - return ok(::paimon::PredicateBuilder::In( - fieldIndex, fieldName, fieldType, literals)); - } + results.resize(firstChild); - // Is null: is_null(field) - if (opName == "is_null") { - return ok( - ::paimon::PredicateBuilder::IsNull(fieldIndex, fieldName, fieldType)); + auto result = opName == "and" ? ::paimon::PredicateBuilder::And(children) + : ::paimon::PredicateBuilder::Or(children); + if (!result.ok()) { + return fail( + opName == "and" ? "PredicateBuilder::And failed" + : "PredicateBuilder::Or failed"); + } + results.push_back(std::move(result).value()); } - // Is not null: is_not_null(field) - if (opName == "is_not_null") { - return ok(::paimon::PredicateBuilder::IsNotNull( - fieldIndex, fieldName, fieldType)); + if (results.size() != 1) { + return fail("predicate translation produced an invalid result stack"); } - - return fail(fmt::format("unsupported opName={}", opName)); + return ok(std::move(results.front())); } std::optional> @@ -684,7 +887,8 @@ PaimonFilterTranslator::extractFieldInfo( std::optional<::paimon::Literal> PaimonFilterTranslator::extractLiteral( const core::TypedExprPtr& expr, - ::paimon::FieldType fieldType) { + ::paimon::FieldType fieldType, + core::ExpressionEvaluator* evaluator) { // Unwrap CastTypedExpr — query planners often wrap constants in casts. auto unwrapped = expr; while (const auto* cast = @@ -695,7 +899,8 @@ std::optional<::paimon::Literal> PaimonFilterTranslator::extractLiteral( const auto* constant = dynamic_cast(unwrapped.get()); if (!constant || constant->hasValueVector()) { - return std::nullopt; + return literalFromVector( + evaluateConstantExpression(unwrapped, evaluator), fieldType); } const auto& value = constant->value(); @@ -945,6 +1150,8 @@ std::optional PaimonFilterTranslator::functionToOpName( return std::string("in"); case ::paimon::Function::Type::NOT_IN: return std::string("not_in"); + case ::paimon::Function::Type::LIKE: + return std::string("like"); default: return std::nullopt; } @@ -1557,157 +1764,188 @@ PaimonFilterTranslator::FilterBuildResult buildFilterFromCall( return fail(fmt::format("unsupported operator '{}'", op)); } -} // namespace - -common::SubfieldFilters PaimonFilterTranslator::toSubfieldFilters( - const core::TypedExprPtr& expr) { - common::SubfieldFilters filters; +void addSubfieldFilter( + common::SubfieldFilters& filters, + common::Subfield subfield, + std::unique_ptr filter) { + if (!filter) { + return; + } + const auto fieldName = subfield.toString(); + auto [it, inserted] = + filters.try_emplace(std::move(subfield), std::move(filter)); + if (!inserted && it->second && filter) { + try { + it->second = it->second->mergeWith(filter.get()); + } catch (const BoltException& e) { + // Some filter types (e.g. HugeintRange) don't support mergeWith. + // Skip pushdown for this subfield rather than failing. + LOG(WARNING) << "toSubfieldFilters: cannot merge filters on '" + << fieldName << "': " << e.message(); + filters.erase(it); + } + } +} - if (!expr) { - return filters; +bool extractSubfieldFilter( + const core::TypedExprPtr& expr, + common::SubfieldFilters& filters) { + const auto* call = dynamic_cast(expr.get()); + if (!call) { + return false; } - // Iterative traversal to satisfy clang-tidy misc-no-recursion. - std::vector stack; - stack.push_back(expr); + const auto& op = call->name(); + + // OR: only push down when ALL direct children are EQUAL predicates + // on the same column (produces a Values filter). Cross-column or mixed + // ORs cannot be pushed into ScanSpec. + if (op == "or") { + std::optional columnName; + std::vector intValues; + std::vector hugeintValues; + std::vector stringValues; + bool allEqual = true; + + std::vector orStack; + for (const auto& child : call->inputs()) { + const auto* childCall = + dynamic_cast(child.get()); + if (childCall) { + orStack.push_back(childCall); + } else { + allEqual = false; + break; + } + } + while (!orStack.empty() && allEqual) { + const auto* current = orStack.back(); + orStack.pop_back(); + if (!current || current->name() != "eq") { + allEqual = false; + break; + } + auto name = extractFieldName(current->inputs()[0]); + if (!name) { + allEqual = false; + break; + } + if (!columnName.has_value()) { + columnName = name; + } else if (*columnName != *name) { + allEqual = false; + break; + } + auto val = extractConstant(current->inputs()[1]); + if (!val) { + allEqual = false; + break; + } + auto kind = current->inputs()[1]->type()->kind(); + if (kind == TypeKind::TINYINT || kind == TypeKind::SMALLINT || + kind == TypeKind::INTEGER || kind == TypeKind::BIGINT) { + // literalToConstantExpr promotes all integral types < BIGINT to + // int64_t in the variant, so always read as int64_t. + intValues.push_back(val->value()); + } else if (kind == TypeKind::HUGEINT) { + hugeintValues.push_back(val->value()); + } else if (kind == TypeKind::VARCHAR || kind == TypeKind::VARBINARY) { + stringValues.push_back(val->value()); + } else { + allEqual = false; + break; + } + } - while (!stack.empty()) { - auto current = std::move(stack.back()); - stack.pop_back(); - if (!current) { - continue; + if (!allEqual) { + LOG(INFO) + << "[FilterPushdown] OR predicate cannot be pushed down: " + "not all children are equality predicates on the same column"; + return false; + } + if (!columnName.has_value()) { + return false; } - const auto* call = dynamic_cast(current.get()); - if (!call) { - continue; + std::unique_ptr filter; + if (!intValues.empty()) { + filter = common::createBigintValues(intValues, false); + } else if (!hugeintValues.empty()) { + auto [minIt, maxIt] = + std::minmax_element(hugeintValues.begin(), hugeintValues.end()); + filter = std::make_unique( + *minIt, *maxIt, hugeintValues, false); + } else if (!stringValues.empty()) { + filter = common::createBytesValues(stringValues, false); + } + if (!filter) { + return false; } + addSubfieldFilter( + filters, common::Subfield(columnName.value()), std::move(filter)); + return true; + } - const auto& op = call->name(); + // Leaf predicate — try to build a filter directly. + auto result = buildFilterFromCall(*call); + if (!result) { + LOG(INFO) << "[FilterPushdown] skipping leaf predicate '" << call->name() + << "': " << result.reason; + return false; + } - // Flatten AND by pushing children onto the stack. - if (op == "and") { - for (const auto& child : call->inputs()) { - stack.push_back(child); - } - continue; - } + auto fieldName = extractFieldName(call->inputs()[0]); + if (!fieldName) { + return false; + } - // OR: only push down when ALL direct children are EQUAL predicates - // on the same column (produces a Values filter). Cross-column or mixed - // ORs cannot be pushed into ScanSpec. - if (op == "or") { - std::optional columnName; - std::vector intValues; - std::vector hugeintValues; - std::vector stringValues; - bool allEqual = true; + addSubfieldFilter( + filters, common::Subfield(*fieldName), std::move(result.value)); + return true; +} - std::vector orStack; - for (const auto& child : call->inputs()) { - const auto* childCall = - dynamic_cast(child.get()); - if (childCall) { - orStack.push_back(childCall); - } else { - allEqual = false; - break; - } - } - while (!orStack.empty() && allEqual) { - const auto* current = orStack.back(); - orStack.pop_back(); - if (!current || current->name() != "eq") { - allEqual = false; - break; - } - auto name = extractFieldName(current->inputs()[0]); - if (!name) { - allEqual = false; - break; - } - if (!columnName.has_value()) { - columnName = name; - } else if (*columnName != *name) { - allEqual = false; - break; - } - auto val = extractConstant(current->inputs()[1]); - if (!val) { - allEqual = false; - break; - } - auto kind = current->inputs()[1]->type()->kind(); - if (kind == TypeKind::TINYINT || kind == TypeKind::SMALLINT || - kind == TypeKind::INTEGER || kind == TypeKind::BIGINT) { - // literalToConstantExpr promotes all integral types < BIGINT to - // int64_t in the variant, so always read as int64_t. - intValues.push_back(val->value()); - } else if (kind == TypeKind::HUGEINT) { - hugeintValues.push_back(val->value()); - } else if (kind == TypeKind::VARCHAR || kind == TypeKind::VARBINARY) { - stringValues.push_back(val->value()); - } else { - allEqual = false; - break; - } - } +bool extractSubfieldFiltersWithEvaluator( + const core::TypedExprPtr& expr, + core::ExpressionEvaluator* evaluator, + common::SubfieldFilters& filters) { + const auto* call = dynamic_cast(expr.get()); + if (!call) { + return false; + } - if (!allEqual) { - LOG(INFO) - << "[FilterPushdown] OR predicate cannot be pushed down: " - "not all children are equality predicates on the same column"; - } - if (allEqual && columnName.has_value()) { - std::unique_ptr filter; - if (!intValues.empty()) { - filter = common::createBigintValues(intValues, false); - } else if (!hugeintValues.empty()) { - auto [minIt, maxIt] = - std::minmax_element(hugeintValues.begin(), hugeintValues.end()); - filter = std::make_unique( - *minIt, *maxIt, hugeintValues, false); - } else if (!stringValues.empty()) { - filter = common::createBytesValues(stringValues, false); - } - if (filter) { - common::Subfield subfield(columnName.value()); - filters.insert_or_assign(std::move(subfield), std::move(filter)); - } - } - continue; - } + try { + auto subfieldFilter = exec::toSubfieldFilter(expr, evaluator); + addSubfieldFilter( + filters, + std::move(subfieldFilter.first), + std::move(subfieldFilter.second)); + return true; + } catch (const BoltException& e) { + LOG(INFO) << "[FilterPushdown] ExprToSubfieldFilterParser skipped '" + << call->name() << "': " << e.message(); + } catch (const std::exception& e) { + LOG(INFO) << "[FilterPushdown] ExprToSubfieldFilterParser skipped '" + << call->name() << "': " << e.what(); + } - // Leaf predicate — try to build a filter directly. - auto result = buildFilterFromCall(*call); - if (!result) { - LOG(INFO) << "[FilterPushdown] skipping leaf predicate '" << call->name() - << "': " << result.reason; - continue; - } - auto filter = std::move(result.value); + return false; +} - auto fieldName = extractFieldName(call->inputs()[0]); - if (!fieldName) { +} // namespace + +common::SubfieldFilters PaimonFilterTranslator::toSubfieldFilters( + const core::TypedExprPtr& expr, + core::ExpressionEvaluator* evaluator) { + common::SubfieldFilters filters; + for (const auto& conjunct : exec::flattenTopLevelConjuncts(expr)) { + if (evaluator && + extractSubfieldFiltersWithEvaluator(conjunct, evaluator, filters)) { continue; } - common::Subfield subfield(*fieldName); - auto [it, inserted] = - filters.try_emplace(std::move(subfield), std::move(filter)); - if (!inserted && it->second && filter) { - try { - it->second = it->second->mergeWith(filter.get()); - } catch (const BoltException& e) { - // Some filter types (e.g. HugeintRange) don't support mergeWith. - // Skip pushdown for this subfield rather than failing. - LOG(WARNING) << "toSubfieldFilters: cannot merge filters on '" - << *fieldName << "': " << e.message(); - filters.erase(it); - } - } + // Fallback to direct subfield filter extraction (no evaluator). + extractSubfieldFilter(conjunct, filters); } - return filters; } diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.h b/bolt/connectors/paimon/PaimonFilterTranslator.h index 0294fd6dd..01609f379 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.h +++ b/bolt/connectors/paimon/PaimonFilterTranslator.h @@ -20,6 +20,7 @@ #include #include #include +#include "bolt/core/ExpressionEvaluator.h" #include "bolt/core/Expressions.h" #include "bolt/core/ITypedExpr.h" #include "bolt/type/filter/FilterBase.h" @@ -62,7 +63,7 @@ struct ToPaimonPredicateResult { /// 2. **Predicate → TypedExpr** (`toTypedExpr`): /// Converts a paimon Predicate back into a bolt TypedExpr expression tree. /// Supported expression patterns: -/// - Comparisons: eq, neq, lt, lte, gt, gte, between, in, not_in +/// - Comparisons: eq, neq, lt, lte, gt, gte, between, in, not_in, like /// - Null checks: is_null, is_not_null /// - Logical: and, or (fully translatable only when all children are) /// - Negation: not (applied per-operator where supported) @@ -78,10 +79,13 @@ class PaimonFilterTranslator { /// @param expr The filter expression tree (typically from TableHandle). /// @param rowType The output row type used to resolve column names to field /// indices required by Paimon's predicate validation. + /// @param evaluator Optional evaluator used to fold constant expressions on + /// the literal side of leaf predicates. /// @return A ToPaimonPredicateResult with the predicate or failure reason. static ToPaimonPredicateResult translate( const core::TypedExprPtr& expr, - const RowTypePtr& rowType); + const RowTypePtr& rowType, + core::ExpressionEvaluator* evaluator = nullptr); // ----------------------------------------------------------------------- // Direction 2a: paimon::Predicate → TypedExprPtr @@ -117,7 +121,7 @@ class PaimonFilterTranslator { memory::MemoryPool* pool = nullptr); // ----------------------------------------------------------------------- - // Direction 2b: TypedExpr → SubfieldFilters (constant-only, no evaluator) + // Direction 2b: TypedExpr → SubfieldFilters // ----------------------------------------------------------------------- /// Result of building a DWIO Filter from a CallTypedExpr. @@ -135,14 +139,18 @@ class PaimonFilterTranslator { /// Convert a TypedExpr into SubfieldFilters for DWIO scan spec pushdown. /// - /// Designed for TypedExpr trees produced by toTypedExpr(), where all - /// literals are ConstantTypedExpr nodes. Does not require an - /// ExpressionEvaluator — values are read directly from constants. + /// When an evaluator is provided, leaf predicates are converted by + /// exec::ExprToSubfieldFilterParser and literal-side constant expressions can + /// be evaluated before building filters. Without an evaluator, this falls + /// back to direct ConstantTypedExpr extraction for TypedExpr trees produced + /// by toTypedExpr(). /// - /// Handles AND-flattening. OR is only converted when all children are - /// EQUAL predicates on the same column (produces a Values filter). + /// Handles AND-flattening. With an evaluator, OR is converted when the + /// common parser can merge both sides on the same subfield. Without an + /// evaluator, OR falls back to the existing same-column equality handling. static common::SubfieldFilters toSubfieldFilters( - const core::TypedExprPtr& expr); + const core::TypedExprPtr& expr, + core::ExpressionEvaluator* evaluator = nullptr); private: struct FieldInfo { @@ -160,7 +168,8 @@ class PaimonFilterTranslator { /// Translate a CallTypedExpr node with row type for field index resolution. static ToPaimonPredicateResult translateCall( const core::CallTypedExpr& call, - const RowTypePtr& rowType); + const RowTypePtr& rowType, + core::ExpressionEvaluator* evaluator); /// Try to extract field reference info, resolving the index from rowType. static std::optional extractFieldInfo( @@ -174,7 +183,8 @@ class PaimonFilterTranslator { /// Extract a constant value as a paimon Literal. static std::optional<::paimon::Literal> extractLiteral( const core::TypedExprPtr& expr, - ::paimon::FieldType fieldType); + ::paimon::FieldType fieldType, + core::ExpressionEvaluator* evaluator = nullptr); /// Extract IN-list literal values from an array ConstantTypedExpr. /// Follows Hive's makeInFilter pattern: uses ArrayVector's diff --git a/bolt/connectors/paimon/PaimonParquetReader.cpp b/bolt/connectors/paimon/PaimonParquetReader.cpp index c10011cfb..bb8307b6f 100644 --- a/bolt/connectors/paimon/PaimonParquetReader.cpp +++ b/bolt/connectors/paimon/PaimonParquetReader.cpp @@ -65,10 +65,12 @@ class PaimonParquetFileBatchReader : public ::paimon::FileBatchReader { std::unique_ptr reader, int32_t batch_size, memory::MemoryPool* const pool, + core::ExpressionEvaluator* expressionEvaluator, uint8_t timestampPrecision) : reader_(std::move(reader)), batch_size_(batch_size), pool_(pool), + expressionEvaluator_(expressionEvaluator), timestampPrecision_(timestampPrecision), readType_(reader_->rowType()) { // LOG(INFO) << "PaimonParquetFileBatchReader created, reader_->rowType() = @@ -131,7 +133,8 @@ class PaimonParquetFileBatchReader : public ::paimon::FileBatchReader { result.ok(), "expression {} not supported for filter pushdown by paimon connector", predicate->ToString()); - auto filters = PaimonFilterTranslator::toSubfieldFilters(result.value); + auto filters = PaimonFilterTranslator::toSubfieldFilters( + result.value, expressionEvaluator_); if (filters.empty()) { LOG(INFO) << "[FilterPushdown] predicate translated successfully but " "produced zero subfield filters — filter will not be " @@ -249,6 +252,7 @@ class PaimonParquetFileBatchReader : public ::paimon::FileBatchReader { std::unique_ptr rowReader_; int32_t batch_size_; memory::MemoryPool* const pool_; + core::ExpressionEvaluator* const expressionEvaluator_; uint8_t timestampPrecision_; RowTypePtr readType_; uint64_t previousBatchFirstRowNumber_{0}; @@ -293,6 +297,7 @@ class PaimonParquetReaderBuilder : public ::paimon::ReaderBuilder { std::move(reader), batch_size_, paimonPool_->getBoltPool(), + paimonPool_->getExpressionEvaluator(), timestampPrecision_); } catch (const std::exception& e) { return ::paimon::Status::IOError( @@ -320,6 +325,7 @@ class PaimonParquetReaderBuilder : public ::paimon::ReaderBuilder { std::move(reader), batch_size_, paimonPool_->getBoltPool(), + paimonPool_->getExpressionEvaluator(), timestampPrecision_); } catch (const std::exception& e) { return ::paimon::Status::IOError( diff --git a/bolt/connectors/paimon/tests/PaimonConnectorTest.cpp b/bolt/connectors/paimon/tests/PaimonConnectorTest.cpp index d5de65272..afe7bb5b2 100644 --- a/bolt/connectors/paimon/tests/PaimonConnectorTest.cpp +++ b/bolt/connectors/paimon/tests/PaimonConnectorTest.cpp @@ -30,6 +30,7 @@ #include "bolt/connectors/paimon/PaimonConfig.h" #include "bolt/connectors/paimon/PaimonConnectorSplit.h" #include "bolt/connectors/paimon/PaimonDataSource.h" +#include "bolt/connectors/paimon/PaimonFilterTranslator.h" #include "bolt/connectors/paimon/PaimonTableHandle.h" #include "bolt/exec/tests/utils/OperatorTestBase.h" #include "bolt/exec/tests/utils/PlanBuilder.h" @@ -888,6 +889,123 @@ TEST_F(PaimonConnectorTest, FilterPushdownIntEquality) { assertQuery(plan, connectorSplits, "SELECT c0 FROM tmp WHERE c0 = 2"); } +TEST_F(PaimonConnectorTest, UnsupportedFilterIsAppliedAsRemainingFilter) { + // Table "basic": id = [1, 2, 3] + // Filter: id + 1 = 3 cannot be translated into a paimon Predicate, but it + // must still be evaluated by the scan as a remaining filter. + auto rootPool = memory::memoryManager()->addRootPool("Test"); + auto leafPool = rootPool->addLeafChild("leaf"); + auto paimonPool = std::make_shared(leafPool.get()); + + auto rowType = ROW({"id"}, {BIGINT()}); + std::unordered_map> + columnHandles; + columnHandles["id"] = std::make_shared("id", BIGINT()); + + std::string tablePath = "file:" + tempDir_->path + "/test_db.db/basic"; + auto filterExpr = parseExpr("id + 1 = 3", rowType); + EXPECT_FALSE(PaimonFilterTranslator::translate(filterExpr, rowType).ok()); + + auto tableHandle = std::make_shared( + "paimon_test", + "test_table", + tablePath, + std::unordered_map(), + filterExpr); + + auto plan = exec::test::PlanBuilder() + .tableScan(rowType, tableHandle, columnHandles) + .planNode(); + + bytedance::bolt::test::VectorMaker mk(leafPool.get()); + auto allRows = mk.rowVector({mk.flatVector({1, 2, 3})}); + createDuckDbTable("tmp", {allRows}); + + auto connectorSplits = makePaimonSplits(tablePath, paimonPool); + assertQuery(plan, connectorSplits, "SELECT c0 FROM tmp WHERE c0 + 1 = 3"); +} + +TEST_F( + PaimonConnectorTest, + UnsupportedFilterCanReadNonOutputColumnAsRemainingFilter) { + // Table "partial_update": ids [1, 2, 3], names [Alice, Bob, Charlie]. + // The scan outputs only name, while the remaining filter references id. + auto rootPool = memory::memoryManager()->addRootPool("Test"); + auto leafPool = rootPool->addLeafChild("leaf"); + auto paimonPool = std::make_shared(leafPool.get()); + + auto filterRowType = ROW({"id", "name"}, {BIGINT(), VARCHAR()}); + auto outputType = ROW({"name"}, {VARCHAR()}); + std::unordered_map> + columnHandles; + columnHandles["name"] = + std::make_shared("name", VARCHAR()); + columnHandles["id"] = std::make_shared("id", BIGINT()); + + std::string tablePath = + "file:" + tempDir_->path + "/test_db.db/partial_update"; + auto filterExpr = parseExpr("id + 1 = 3", filterRowType); + EXPECT_FALSE( + PaimonFilterTranslator::translate(filterExpr, filterRowType).ok()); + + auto tableHandle = std::make_shared( + "paimon_test", + "test_table", + tablePath, + std::unordered_map(), + filterExpr); + + auto plan = exec::test::PlanBuilder() + .tableScan(outputType, tableHandle, columnHandles) + .planNode(); + + bytedance::bolt::test::VectorMaker mk(leafPool.get()); + auto allRows = mk.rowVector( + {mk.flatVector({1, 2, 3}), + mk.flatVector({"Alice", "Bob", "Charlie"})}); + createDuckDbTable("tmp", {allRows}); + + auto connectorSplits = makePaimonSplits(tablePath, paimonPool); + assertQuery(plan, connectorSplits, "SELECT c1 FROM tmp WHERE c0 + 1 = 3"); +} + +TEST_F(PaimonConnectorTest, PartiallyTranslatableFilterKeepsRemainingFilter) { + auto rootPool = memory::memoryManager()->addRootPool("Test"); + auto leafPool = rootPool->addLeafChild("leaf"); + auto paimonPool = std::make_shared(leafPool.get()); + + auto rowType = ROW({"id"}, {BIGINT()}); + std::unordered_map> + columnHandles; + columnHandles["id"] = std::make_shared("id", BIGINT()); + + std::string tablePath = "file:" + tempDir_->path + "/test_db.db/basic"; + auto filterExpr = parseExpr("id > 1 AND id + 1 = 3", rowType); + EXPECT_FALSE(PaimonFilterTranslator::translate(filterExpr, rowType).ok()); + EXPECT_TRUE( + PaimonFilterTranslator::translate(parseExpr("id > 1", rowType), rowType) + .ok()); + + auto tableHandle = std::make_shared( + "paimon_test", + "test_table", + tablePath, + std::unordered_map(), + filterExpr); + + auto plan = exec::test::PlanBuilder() + .tableScan(rowType, tableHandle, columnHandles) + .planNode(); + + bytedance::bolt::test::VectorMaker mk(leafPool.get()); + auto allRows = mk.rowVector({mk.flatVector({1, 2, 3})}); + createDuckDbTable("tmp", {allRows}); + + auto connectorSplits = makePaimonSplits(tablePath, paimonPool); + assertQuery( + plan, connectorSplits, "SELECT c0 FROM tmp WHERE c0 > 1 AND c0 + 1 = 3"); +} + TEST_F(PaimonConnectorTest, FilterPushdownIntGreaterThan) { // Table "basic": id = [1, 2, 3] // Filter: id > 1 → expect [2, 3] diff --git a/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp index 2ddd77912..146e768a7 100644 --- a/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp +++ b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp @@ -27,6 +27,11 @@ #include "bolt/common/memory/Memory.h" #include "bolt/connectors/paimon/PaimonFilterTranslator.h" #include "bolt/core/Expressions.h" +#include "bolt/core/QueryCtx.h" +#include "bolt/expression/Expr.h" +#include "bolt/expression/ExprToSubfieldFilter.h" +#include "bolt/functions/prestosql/registration/RegistrationFunctions.h" +#include "bolt/type/Subfield.h" #include "bolt/type/Type.h" #include "bolt/vector/BaseVector.h" #include "bolt/vector/tests/utils/VectorTestBase.h" @@ -42,6 +47,7 @@ class PaimonFilterTranslatorTest public bytedance::bolt::test::VectorTestBase { protected: static void SetUpTestSuite() { + functions::prestosql::registerAllScalarFunctions(); memory::MemoryManager::testingSetInstance(memory::MemoryManager::Options{}); } @@ -60,6 +66,11 @@ class PaimonFilterTranslatorTest return PaimonFilterTranslator::translate(expr, rowType_); } + ToPaimonPredicateResult translateWithEvaluator( + const core::TypedExprPtr& expr) { + return PaimonFilterTranslator::translate(expr, rowType_, &evaluator_); + } + const RowTypePtr rowType_{ROW({ {"col", BIGINT()}, {"id", BIGINT()}, @@ -98,6 +109,10 @@ class PaimonFilterTranslatorTest return std::make_shared(VARCHAR(), variant(value)); } + static TypedExprPtr binaryConst(const std::string& value) { + return std::make_shared(VARBINARY(), variant(value)); + } + static TypedExprPtr floatConst(float value) { return std::make_shared(REAL(), value); } @@ -170,6 +185,15 @@ class PaimonFilterTranslatorTest BOOLEAN(), std::vector{lhs, rhs}, "eq"); } + static TypedExprPtr plus(const TypedExprPtr& lhs, const TypedExprPtr& rhs) { + return std::make_shared( + BIGINT(), std::vector{lhs, rhs}, "plus"); + } + + static TypedExprPtr concat(const std::vector& inputs) { + return std::make_shared(VARCHAR(), inputs, "concat"); + } + static TypedExprPtr neq(const TypedExprPtr& lhs, const TypedExprPtr& rhs) { return std::make_shared( BOOLEAN(), std::vector{lhs, rhs}, "neq"); @@ -227,6 +251,13 @@ class PaimonFilterTranslatorTest BOOLEAN(), std::vector{col}, "is_not_null"); } + static TypedExprPtr likeExpr( + const TypedExprPtr& col, + const TypedExprPtr& pattern) { + return std::make_shared( + BOOLEAN(), std::vector{col, pattern}, "like"); + } + static TypedExprPtr notExpr(const TypedExprPtr& inner) { return std::make_shared( BOOLEAN(), std::vector{inner}, "not"); @@ -245,6 +276,9 @@ class PaimonFilterTranslatorTest return std::make_shared( BOOLEAN(), std::vector{left, right}, "or"); } + + std::shared_ptr queryCtx_{core::QueryCtx::create()}; + exec::SimpleExpressionEvaluator evaluator_{queryCtx_.get(), pool_.get()}; }; // --------------------------------------------------------------------------- @@ -268,6 +302,20 @@ TEST_F(PaimonFilterTranslatorTest, BareConstantReturnsNull) { EXPECT_FALSE(result.ok()) << result.reason; } +TEST_F(PaimonFilterTranslatorTest, FlattenTopLevelConjunctsKeepsLeafOrder) { + auto first = eq(field(BIGINT(), "id"), intConst(1)); + auto second = gt(field(BIGINT(), "age"), intConst(10)); + auto third = isNotNull(field(VARCHAR(), "name")); + auto expr = andExpr(andExpr(first, second), third); + + auto conjuncts = exec::flattenTopLevelConjuncts(expr); + + ASSERT_EQ(conjuncts.size(), 3); + EXPECT_EQ(conjuncts[0].get(), first.get()); + EXPECT_EQ(conjuncts[1].get(), second.get()); + EXPECT_EQ(conjuncts[2].get(), third.get()); +} + // --------------------------------------------------------------------------- // Equality predicates // --------------------------------------------------------------------------- @@ -295,6 +343,162 @@ TEST_F(PaimonFilterTranslatorTest, StringEquality) { EXPECT_NE(pred.value->ToString().find("alice"), std::string::npos); } +TEST_F(PaimonFilterTranslatorTest, EvaluatesConstantRhsExpression) { + auto expr = eq(field(BIGINT(), "id"), plus(intConst(1), intConst(1))); + + auto withoutEvaluator = translate(expr); + EXPECT_FALSE(withoutEvaluator.ok()) << withoutEvaluator.reason; + + auto pred = translateWithEvaluator(expr); + ASSERT_TRUE(pred.ok()) << pred.reason; + const auto* leaf = + dynamic_cast(pred.value.get()); + ASSERT_NE(leaf, nullptr); + EXPECT_EQ(leaf->GetFunction().GetType(), ::paimon::Function::Type::EQUAL); + EXPECT_EQ(leaf->FieldName(), "id"); + ASSERT_EQ(leaf->Literals().size(), 1); + EXPECT_EQ(leaf->Literals()[0].GetValue(), 2); +} + +TEST_F(PaimonFilterTranslatorTest, StringLike) { + auto expr = likeExpr(field(VARCHAR(), "name"), strConst("ali%")); + auto pred = translate(expr); + ASSERT_TRUE(pred.ok()) << pred.reason; + + const auto* leaf = + dynamic_cast(pred.value.get()); + ASSERT_NE(leaf, nullptr); + EXPECT_EQ(leaf->GetFunction().GetType(), ::paimon::Function::Type::LIKE); + EXPECT_EQ(leaf->FieldName(), "name"); + ASSERT_EQ(leaf->Literals().size(), 1); + EXPECT_EQ(leaf->Literals()[0].GetValue(), "ali%"); +} + +TEST_F(PaimonFilterTranslatorTest, StringLikeEvaluatesConstantPattern) { + auto expr = likeExpr( + field(VARCHAR(), "name"), concat({strConst("ali"), strConst("%")})); + + auto withoutEvaluator = translate(expr); + EXPECT_FALSE(withoutEvaluator.ok()) << withoutEvaluator.reason; + + auto pred = translateWithEvaluator(expr); + ASSERT_TRUE(pred.ok()) << pred.reason; + const auto* leaf = + dynamic_cast(pred.value.get()); + ASSERT_NE(leaf, nullptr); + EXPECT_EQ(leaf->GetFunction().GetType(), ::paimon::Function::Type::LIKE); + ASSERT_EQ(leaf->Literals().size(), 1); + EXPECT_EQ(leaf->Literals()[0].GetValue(), "ali%"); +} + +TEST_F(PaimonFilterTranslatorTest, StringLikeWithEscapeStaysRemaining) { + auto expr = std::make_shared( + BOOLEAN(), + std::vector{ + field(VARCHAR(), "name"), strConst("ali\\_%"), strConst("\\")}, + "like"); + + auto pred = translateWithEvaluator(expr); + EXPECT_FALSE(pred.ok()) << pred.reason; +} + +TEST_F(PaimonFilterTranslatorTest, EvaluatesConstantRhsRangeExpression) { + auto expr = gte(field(BIGINT(), "id"), plus(intConst(10), intConst(5))); + + auto pred = translateWithEvaluator(expr); + ASSERT_TRUE(pred.ok()) << pred.reason; + const auto* leaf = + dynamic_cast(pred.value.get()); + ASSERT_NE(leaf, nullptr); + EXPECT_EQ( + leaf->GetFunction().GetType(), + ::paimon::Function::Type::GREATER_OR_EQUAL); + EXPECT_EQ(leaf->FieldName(), "id"); + ASSERT_EQ(leaf->Literals().size(), 1); + EXPECT_EQ(leaf->Literals()[0].GetValue(), 15); +} + +TEST_F(PaimonFilterTranslatorTest, EvaluatesConstantRhsBetweenExpressions) { + auto expr = between( + field(BIGINT(), "val"), + plus(intConst(1), intConst(2)), + plus(intConst(10), intConst(5))); + + auto pred = translateWithEvaluator(expr); + ASSERT_TRUE(pred.ok()) << pred.reason; + const auto typed = + PaimonFilterTranslator::toTypedExpr(pred.value, pool_.get()); + ASSERT_TRUE(typed.ok()) << typed.reason; + + auto filters = PaimonFilterTranslator::toSubfieldFilters(typed.value); + ASSERT_EQ(filters.size(), 1); + auto it = filters.find(common::Subfield("val")); + ASSERT_NE(it, filters.end()); + EXPECT_TRUE(it->second->testInt64(3)); + EXPECT_TRUE(it->second->testInt64(15)); + EXPECT_FALSE(it->second->testInt64(2)); + EXPECT_FALSE(it->second->testInt64(16)); +} + +TEST_F(PaimonFilterTranslatorTest, NonConstantRhsExpressionIsNotPushedDown) { + auto expr = + eq(field(BIGINT(), "id"), plus(field(BIGINT(), "a"), intConst(1))); + + auto pred = translateWithEvaluator(expr); + EXPECT_FALSE(pred.ok()) << pred.reason; +} + +TEST_F( + PaimonFilterTranslatorTest, + EvaluatedConstantRhsRoundTripsToSubfieldFilter) { + auto expr = eq(field(BIGINT(), "id"), plus(intConst(1), intConst(1))); + auto pred = translateWithEvaluator(expr); + ASSERT_TRUE(pred.ok()) << pred.reason; + + auto typed = PaimonFilterTranslator::toTypedExpr(pred.value, pool_.get()); + ASSERT_TRUE(typed.ok()) << typed.reason; + const auto* call = + dynamic_cast(typed.value.get()); + ASSERT_NE(call, nullptr); + const auto* constant = + dynamic_cast(call->inputs()[1].get()); + ASSERT_NE(constant, nullptr); + EXPECT_EQ(constant->value().value(), 2); + + auto filters = PaimonFilterTranslator::toSubfieldFilters(typed.value); + ASSERT_EQ(filters.size(), 1); + auto it = filters.find(common::Subfield("id")); + ASSERT_NE(it, filters.end()); + EXPECT_TRUE(it->second->testInt64(2)); + EXPECT_FALSE(it->second->testInt64(1)); +} + +TEST_F(PaimonFilterTranslatorTest, SubfieldFiltersUseEvaluatorForConstantRhs) { + auto expr = eq(field(BIGINT(), "id"), plus(intConst(1), intConst(1))); + + EXPECT_TRUE(PaimonFilterTranslator::toSubfieldFilters(expr).empty()); + + auto filters = PaimonFilterTranslator::toSubfieldFilters(expr, &evaluator_); + ASSERT_EQ(filters.size(), 1); + auto it = filters.find(common::Subfield("id")); + ASSERT_NE(it, filters.end()); + EXPECT_TRUE(it->second->testInt64(2)); + EXPECT_FALSE(it->second->testInt64(1)); +} + +TEST_F( + PaimonFilterTranslatorTest, + SubfieldFiltersFallbackWhenEvaluatorParserDoesNotSupportType) { + auto expr = eq(field(VARBINARY(), "payload"), binaryConst("abc")); + + auto filters = PaimonFilterTranslator::toSubfieldFilters(expr, &evaluator_); + ASSERT_EQ(filters.size(), 1); + auto it = filters.find(common::Subfield("payload")); + ASSERT_NE(it, filters.end()); + EXPECT_TRUE(it->second->testBytes("abc", 3)); + EXPECT_FALSE(it->second->testBytes("bcd", 3)); +} + // --------------------------------------------------------------------------- // Range comparison predicates // --------------------------------------------------------------------------- diff --git a/bolt/dwio/parquet/reader/ParquetReader.cpp b/bolt/dwio/parquet/reader/ParquetReader.cpp index ea79651c9..62a1d5527 100644 --- a/bolt/dwio/parquet/reader/ParquetReader.cpp +++ b/bolt/dwio/parquet/reader/ParquetReader.cpp @@ -2024,6 +2024,8 @@ ParquetRowReader::ParquetRowReader( impl_ = std::make_unique(readerBase, options); } +ParquetRowReader::~ParquetRowReader() = default; + void ParquetRowReader::filterRowGroups() { impl_->filterRowGroups(); } diff --git a/bolt/dwio/parquet/reader/ParquetReader.h b/bolt/dwio/parquet/reader/ParquetReader.h index fc2822f96..157f9d4b3 100644 --- a/bolt/dwio/parquet/reader/ParquetReader.h +++ b/bolt/dwio/parquet/reader/ParquetReader.h @@ -60,7 +60,7 @@ class ParquetRowReader : public dwio::common::RowReader { ParquetRowReader( const std::shared_ptr& readerBase, const dwio::common::RowReaderOptions& options); - ~ParquetRowReader() override = default; + ~ParquetRowReader() override; int64_t nextRowNumber() override; diff --git a/bolt/expression/ExprToSubfieldFilter.cpp b/bolt/expression/ExprToSubfieldFilter.cpp index 978f13714..9de40ba86 100644 --- a/bolt/expression/ExprToSubfieldFilter.cpp +++ b/bolt/expression/ExprToSubfieldFilter.cpp @@ -135,6 +135,41 @@ std::pair> toSubfieldFilter( "Unsupported expression for range filter: {}", expr->toString()); } +void flattenTopLevelConjuncts( + const core::TypedExprPtr& expr, + std::vector& conjuncts) { + std::vector pending; + if (expr) { + pending.push_back(expr); + } + + while (!pending.empty()) { + auto current = std::move(pending.back()); + pending.pop_back(); + if (!current) { + continue; + } + auto call = std::dynamic_pointer_cast(current); + if (call && call->name() == "and") { + const auto& inputs = call->inputs(); + for (size_t i = inputs.size(); i > 0; --i) { + pending.push_back(inputs[i - 1]); + } + continue; + } + conjuncts.push_back(std::move(current)); + } +} + +std::vector flattenTopLevelConjuncts( + const core::TypedExprPtr& expr) { + std::vector conjuncts; + if (expr) { + flattenTopLevelConjuncts(expr, conjuncts); + } + return conjuncts; +} + std::shared_ptr ExprToSubfieldFilterParser::parser_ = std::make_shared(); diff --git a/bolt/expression/ExprToSubfieldFilter.h b/bolt/expression/ExprToSubfieldFilter.h index 7e5b5db39..41e697e16 100644 --- a/bolt/expression/ExprToSubfieldFilter.h +++ b/bolt/expression/ExprToSubfieldFilter.h @@ -30,6 +30,7 @@ #pragma once +#include #include "bolt/core/ExpressionEvaluator.h" #include "bolt/core/Expressions.h" #include "bolt/core/ITypedExpr.h" @@ -422,6 +423,17 @@ std::pair> toSubfieldFilter( const core::TypedExprPtr& expr, core::ExpressionEvaluator* evaluator); +/// Flattens top-level AND expressions into a list of conjuncts. +/// +/// This only expands AND calls. Other expressions, including OR and NOT, are +/// returned as leaf conjuncts. +std::vector flattenTopLevelConjuncts( + const core::TypedExprPtr& expr); + +void flattenTopLevelConjuncts( + const core::TypedExprPtr& expr, + std::vector& conjuncts); + /// Language-specific translator of generic expressions into subfield filters. /// Default language is Presto. A different language can be supported by /// registering language-specific implementation using 'registerParser' API.