From 9cc6d69e726aeb2ec72d3da9152a74978f82b4fb Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Wed, 24 Jun 2026 17:18:14 +0800 Subject: [PATCH 1/4] fix: improve Paimon filter pushdown Preserve untranslated Paimon filters as remaining filters, fold constant RHS expressions before predicate translation, and support string LIKE predicates. Add translator coverage for remaining filters, constant RHS pushdown, subfield filter round trips, and string LIKE cases. --- bolt/connectors/paimon/PaimonDataSource.cpp | 284 ++++++++++++------ bolt/connectors/paimon/PaimonDataSource.h | 11 + .../paimon/PaimonFilterTranslator.cpp | 186 +++++++++++- .../paimon/PaimonFilterTranslator.h | 14 +- .../paimon/tests/PaimonConnectorTest.cpp | 118 ++++++++ .../tests/PaimonFilterTranslatorTest.cpp | 159 ++++++++++ 6 files changed, 662 insertions(+), 110 deletions(-) diff --git a/bolt/connectors/paimon/PaimonDataSource.cpp b/bolt/connectors/paimon/PaimonDataSource.cpp index 125fd75ea..fe591e58e 100644 --- a/bolt/connectors/paimon/PaimonDataSource.cpp +++ b/bolt/connectors/paimon/PaimonDataSource.cpp @@ -18,17 +18,21 @@ #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/type/StringView.h" #include "bolt/type/Type.h" #include "bolt/vector/BaseVector.h" @@ -38,6 +42,126 @@ #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 { + +void collectTopLevelConjuncts( + const core::TypedExprPtr& expression, + std::vector& conjuncts) { + const auto* call = dynamic_cast(expression.get()); + if (call && call->name() == "and") { + for (const auto& input : call->inputs()) { + collectTopLevelConjuncts(input, conjuncts); + } + return; + } + conjuncts.push_back(expression); +} + +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) { + if (const auto* field = + dynamic_cast(expression.get())) { + if (field->inputs().empty() && seen.insert(field->name()).second) { + fieldNames.push_back(field->name()); + } + } + + for (const auto& input : expression->inputs()) { + collectFieldNames(input, fieldNames, seen); + } +} + +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; + collectTopLevelConjuncts(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,6 +173,7 @@ 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_); @@ -56,14 +181,38 @@ PaimonDataSource::PaimonDataSource( ::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 +275,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 +397,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 +420,50 @@ 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..60fc5f0a6 100644 --- a/bolt/connectors/paimon/PaimonDataSource.h +++ b/bolt/connectors/paimon/PaimonDataSource.h @@ -20,6 +20,8 @@ #include "bolt/connectors/paimon/PaimonConfig.h" #include "bolt/connectors/paimon/PaimonConnectorSplit.h" #include "bolt/connectors/paimon/PaimonTableHandle.h" +#include "bolt/exec/OperatorUtils.h" +#include "bolt/vector/SelectivityVector.h" // Forward declare paimon types namespace paimon { @@ -64,8 +66,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 +82,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..0f6dde5eb 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.cpp +++ b/bolt/connectors/paimon/PaimonFilterTranslator.cpp @@ -28,6 +28,7 @@ #include #include #include "bolt/core/Expressions.h" +#include "bolt/expression/Expr.h" #include "bolt/type/Subfield.h" #include "bolt/type/Timestamp.h" #include "bolt/type/filter/FilterBase.h" @@ -188,6 +189,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; + int32_t nanoOfMillis = static_cast(totalNanos % 1'000'000); + return ::paimon::Literal(::paimon::Timestamp(millis, nanoOfMillis)); + } + default: + return std::nullopt; + } +} + } // namespace // =========================================================================== @@ -224,7 +350,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 +366,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)}; @@ -262,7 +390,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( std::vector> children; children.reserve(inputs.size()); for (const auto& input : inputs) { - auto child = translate(input, rowType); + auto child = translate(input, rowType, evaluator); if (!child.ok()) { return fail(fmt::format("AND child failed: {}", child.reason)); } @@ -280,7 +408,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( std::vector> children; children.reserve(inputs.size()); for (const auto& input : inputs) { - auto child = translate(input, rowType); + auto child = translate(input, rowType, evaluator); if (!child.ok()) { return fail("OR child failed"); } @@ -307,7 +435,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (negatedOp.has_value()) { core::CallTypedExpr negatedCall( innerCall->type(), innerCall->inputs(), negatedOp.value()); - return translateCall(negatedCall, rowType); + return translateCall(negatedCall, rowType, evaluator); } if (normalizeOpName(innerCall->name()) == "is_null") { auto fieldInfo = extractFieldInfo(innerCall->inputs()[0], rowType); @@ -350,7 +478,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 2) { return fail("eq requires 2 inputs"); } - auto lit = extractLiteral(inputs[1], fieldType); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); if (!lit) { return fail("eq: could not extract literal"); } @@ -363,7 +491,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 2) { return fail("neq requires 2 inputs"); } - auto lit = extractLiteral(inputs[1], fieldType); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); if (!lit) { return fail("neq: could not extract literal"); } @@ -376,7 +504,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 2) { return fail("lt requires 2 inputs"); } - auto lit = extractLiteral(inputs[1], fieldType); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); if (!lit) { return fail("lt: could not extract literal"); } @@ -389,7 +517,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 2) { return fail("lte requires 2 inputs"); } - auto lit = extractLiteral(inputs[1], fieldType); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); if (!lit) { return fail("lte: could not extract literal"); } @@ -402,7 +530,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 2) { return fail("gt requires 2 inputs"); } - auto lit = extractLiteral(inputs[1], fieldType); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); if (!lit) { return fail("gt: could not extract literal"); } @@ -415,7 +543,7 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 2) { return fail("gte requires 2 inputs"); } - auto lit = extractLiteral(inputs[1], fieldType); + auto lit = extractLiteral(inputs[1], fieldType, evaluator); if (!lit) { return fail("gte: could not extract literal"); } @@ -428,8 +556,8 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( if (inputs.size() < 3) { return fail("between requires 3 inputs"); } - auto low = extractLiteral(inputs[1], fieldType); - auto high = extractLiteral(inputs[2], fieldType); + auto low = extractLiteral(inputs[1], fieldType, evaluator); + auto high = extractLiteral(inputs[2], fieldType, evaluator); if (!low || !high) { return fail("between: could not extract literals"); } @@ -453,6 +581,28 @@ ToPaimonPredicateResult PaimonFilterTranslator::translateCall( fieldIndex, fieldName, fieldType, literals)); } + // 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()); + } + // Is null: is_null(field) if (opName == "is_null") { return ok( @@ -684,7 +834,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 +846,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 +1097,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; } diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.h b/bolt/connectors/paimon/PaimonFilterTranslator.h index 0294fd6dd..81e509daf 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 @@ -160,7 +164,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 +179,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/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..eb6d8543f 100644 --- a/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp +++ b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp @@ -27,6 +27,10 @@ #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/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 +46,7 @@ class PaimonFilterTranslatorTest public bytedance::bolt::test::VectorTestBase { protected: static void SetUpTestSuite() { + functions::prestosql::registerAllScalarFunctions(); memory::MemoryManager::testingSetInstance(memory::MemoryManager::Options{}); } @@ -60,6 +65,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()}, @@ -170,6 +180,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 +246,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 +271,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()}; }; // --------------------------------------------------------------------------- @@ -295,6 +324,136 @@ 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)); +} + // --------------------------------------------------------------------------- // Range comparison predicates // --------------------------------------------------------------------------- From 867e4ce82c7969dc1b0cff278f7e6ceb8338af01 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Tue, 30 Jun 2026 20:23:54 +0800 Subject: [PATCH 2/4] fix: refactor Paimon subfield filter extraction Reuse the common ExprToSubfieldFilter parser when an evaluator is available and keep the Paimon direct filter path as fallback. Propagate the evaluator through the Paimon scan path and move top-level conjunction flattening into the shared expression filter utilities. --- bolt/connectors/paimon/BoltMemoryPool.h | 12 +- bolt/connectors/paimon/PaimonDataSource.cpp | 25 +- bolt/connectors/paimon/PaimonDataSource.h | 1 + .../paimon/PaimonFilterTranslator.cpp | 308 ++++++++++-------- .../paimon/PaimonFilterTranslator.h | 18 +- .../connectors/paimon/PaimonParquetReader.cpp | 8 +- .../tests/PaimonFilterTranslatorTest.cpp | 45 +++ bolt/expression/ExprToSubfieldFilter.cpp | 25 ++ bolt/expression/ExprToSubfieldFilter.h | 12 + 9 files changed, 285 insertions(+), 169 deletions(-) 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 fe591e58e..9a18e03d2 100644 --- a/bolt/connectors/paimon/PaimonDataSource.cpp +++ b/bolt/connectors/paimon/PaimonDataSource.cpp @@ -33,6 +33,7 @@ #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" @@ -77,19 +78,6 @@ std::shared_ptr findPaimonColumnHandle( namespace { -void collectTopLevelConjuncts( - const core::TypedExprPtr& expression, - std::vector& conjuncts) { - const auto* call = dynamic_cast(expression.get()); - if (call && call->name() == "and") { - for (const auto& input : call->inputs()) { - collectTopLevelConjuncts(input, conjuncts); - } - return; - } - conjuncts.push_back(expression); -} - core::TypedExprPtr combineConjuncts(std::vector conjuncts) { if (conjuncts.empty()) { return nullptr; @@ -130,7 +118,7 @@ std::pair, core::TypedExprPtr> planFilter( const RowTypePtr& rowType, core::ExpressionEvaluator* evaluator) { std::vector conjuncts; - collectTopLevelConjuncts(expression, conjuncts); + exec::flattenTopLevelConjuncts(expression, conjuncts); std::vector> predicates; std::vector remainingConjuncts; @@ -176,7 +164,8 @@ PaimonDataSource::PaimonDataSource( 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; @@ -433,11 +422,7 @@ std::optional PaimonDataSource::next( BufferPtr remainingIndices; if (remainingFilterExprSet_) { auto filterInput = std::make_shared( - pool_, - filterRowType_, - row->nulls(), - row->size(), - row->children()); + pool_, filterRowType_, row->nulls(), row->size(), row->children()); rowsRemaining = evaluateRemainingFilter(filterInput); if (rowsRemaining == 0) { return RowVector::createEmpty(outputType_, pool_); diff --git a/bolt/connectors/paimon/PaimonDataSource.h b/bolt/connectors/paimon/PaimonDataSource.h index 60fc5f0a6..4d3fe1cec 100644 --- a/bolt/connectors/paimon/PaimonDataSource.h +++ b/bolt/connectors/paimon/PaimonDataSource.h @@ -20,6 +20,7 @@ #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" diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.cpp b/bolt/connectors/paimon/PaimonFilterTranslator.cpp index 0f6dde5eb..a25a6b128 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.cpp +++ b/bolt/connectors/paimon/PaimonFilterTranslator.cpp @@ -29,6 +29,7 @@ #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" @@ -1711,157 +1712,186 @@ PaimonFilterTranslator::FilterBuildResult buildFilterFromCall( return fail(fmt::format("unsupported operator '{}'", op)); } -} // namespace - -common::SubfieldFilters PaimonFilterTranslator::toSubfieldFilters( - const core::TypedExprPtr& expr) { - common::SubfieldFilters filters; - - if (!expr) { - return filters; - } - - // Iterative traversal to satisfy clang-tidy misc-no-recursion. - std::vector stack; - stack.push_back(expr); - - while (!stack.empty()) { - auto current = std::move(stack.back()); - stack.pop_back(); - if (!current) { - continue; - } - - const auto* call = dynamic_cast(current.get()); - if (!call) { - continue; +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); } + } +} - const auto& op = call->name(); +bool extractSubfieldFilter( + const core::TypedExprPtr& expr, + common::SubfieldFilters& filters) { + const auto* call = dynamic_cast(expr.get()); + if (!call) { + return false; + } - // Flatten AND by pushing children onto the stack. - if (op == "and") { - for (const auto& child : call->inputs()) { - stack.push_back(child); + 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; } - continue; - } - - // 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; } - 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; - } + auto name = extractFieldName(current->inputs()[0]); + if (!name) { + allEqual = false; + break; } - - if (!allEqual) { - LOG(INFO) - << "[FilterPushdown] OR predicate cannot be pushed down: " - "not all children are equality predicates on the same column"; + if (!columnName.has_value()) { + columnName = name; + } else if (*columnName != *name) { + allEqual = false; + break; } - 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)); - } + auto val = extractConstant(current->inputs()[1]); + if (!val) { + allEqual = false; + break; } - continue; - } - - // 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); - - auto fieldName = extractFieldName(call->inputs()[0]); - if (!fieldName) { - 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); + 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; } } + + 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; + } + + 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; + } + + // 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; + } + + auto fieldName = extractFieldName(call->inputs()[0]); + if (!fieldName) { + return false; + } + + addSubfieldFilter( + filters, common::Subfield(*fieldName), std::move(result.value)); + return true; +} + +bool extractSubfieldFiltersWithEvaluator( + const core::TypedExprPtr& expr, + core::ExpressionEvaluator* evaluator, + common::SubfieldFilters& filters) { + const auto* call = dynamic_cast(expr.get()); + if (!call) { + return false; + } + + 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(); + } + + return false; +} + +} // namespace + +common::SubfieldFilters PaimonFilterTranslator::toSubfieldFilters( + const core::TypedExprPtr& expr, + core::ExpressionEvaluator* evaluator) { + common::SubfieldFilters filters; + if (evaluator && + extractSubfieldFiltersWithEvaluator(expr, evaluator, filters)) { + return filters; } + // fallback to direct subfield filter extraction (no evaluator) + extractSubfieldFilter(expr, filters); return filters; } diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.h b/bolt/connectors/paimon/PaimonFilterTranslator.h index 81e509daf..01609f379 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.h +++ b/bolt/connectors/paimon/PaimonFilterTranslator.h @@ -121,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. @@ -139,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 { 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/PaimonFilterTranslatorTest.cpp b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp index eb6d8543f..bb6ffdf3c 100644 --- a/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp +++ b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp @@ -29,6 +29,7 @@ #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" @@ -108,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); } @@ -297,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 // --------------------------------------------------------------------------- @@ -454,6 +473,32 @@ TEST_F( 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("abd", 3)); +} + // --------------------------------------------------------------------------- // Range comparison predicates // --------------------------------------------------------------------------- diff --git a/bolt/expression/ExprToSubfieldFilter.cpp b/bolt/expression/ExprToSubfieldFilter.cpp index 978f13714..70f21b153 100644 --- a/bolt/expression/ExprToSubfieldFilter.cpp +++ b/bolt/expression/ExprToSubfieldFilter.cpp @@ -135,6 +135,31 @@ std::pair> toSubfieldFilter( "Unsupported expression for range filter: {}", expr->toString()); } +void flattenTopLevelConjuncts( + const core::TypedExprPtr& expr, + std::vector& conjuncts) { + if (!expr) { + return; + } + auto call = std::dynamic_pointer_cast(expr); + if (call && call->name() == "and") { + for (const auto& input : call->inputs()) { + flattenTopLevelConjuncts(input, conjuncts); + } + return; + } + conjuncts.push_back(expr); +} + +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. From 13a62992284e76258567e64504de7a594b4150b7 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Thu, 2 Jul 2026 17:44:00 +0800 Subject: [PATCH 3/4] fix(paimon): preserve remaining filter pushdown inputs Collect top-level field accesses backed by InputTypedExpr so remaining filters can read non-output columns during Paimon scans. Flatten top-level conjuncts before converting Paimon predicates to subfield filters so evaluator-backed extraction can push down each supported leaf. --- bolt/connectors/paimon/PaimonDataSource.cpp | 13 ++++++++----- bolt/connectors/paimon/PaimonFilterTranslator.cpp | 14 ++++++++------ 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/bolt/connectors/paimon/PaimonDataSource.cpp b/bolt/connectors/paimon/PaimonDataSource.cpp index 9a18e03d2..23f569f2a 100644 --- a/bolt/connectors/paimon/PaimonDataSource.cpp +++ b/bolt/connectors/paimon/PaimonDataSource.cpp @@ -93,11 +93,14 @@ void collectFieldNames( const core::TypedExprPtr& expression, std::vector& fieldNames, std::unordered_set& seen) { - if (const auto* field = - dynamic_cast(expression.get())) { - if (field->inputs().empty() && seen.insert(field->name()).second) { - fieldNames.push_back(field->name()); - } + const auto* field = + dynamic_cast(expression.get()); + if (field && + (field->inputs().empty() || + dynamic_cast( + field->inputs().front().get()) != nullptr) && + seen.insert(field->name()).second) { + fieldNames.push_back(field->name()); } for (const auto& input : expression->inputs()) { diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.cpp b/bolt/connectors/paimon/PaimonFilterTranslator.cpp index a25a6b128..db5ef1c3b 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.cpp +++ b/bolt/connectors/paimon/PaimonFilterTranslator.cpp @@ -1885,13 +1885,15 @@ common::SubfieldFilters PaimonFilterTranslator::toSubfieldFilters( const core::TypedExprPtr& expr, core::ExpressionEvaluator* evaluator) { common::SubfieldFilters filters; - if (evaluator && - extractSubfieldFiltersWithEvaluator(expr, evaluator, filters)) { - return filters; - } + for (const auto& conjunct : exec::flattenTopLevelConjuncts(expr)) { + if (evaluator && + extractSubfieldFiltersWithEvaluator(conjunct, evaluator, filters)) { + continue; + } - // fallback to direct subfield filter extraction (no evaluator) - extractSubfieldFilter(expr, filters); + // Fallback to direct subfield filter extraction (no evaluator). + extractSubfieldFilter(conjunct, filters); + } return filters; } From 28d61e078be7ac84164aa2594b404a089071d25a Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Thu, 2 Jul 2026 20:35:12 +0800 Subject: [PATCH 4/4] fix recursive call chain --- bolt/connectors/paimon/PaimonDataSource.cpp | 34 +- .../paimon/PaimonFilterTranslator.cpp | 446 ++++++++++-------- .../tests/PaimonFilterTranslatorTest.cpp | 2 +- bolt/dwio/parquet/reader/ParquetReader.cpp | 2 + bolt/dwio/parquet/reader/ParquetReader.h | 2 +- bolt/expression/ExprToSubfieldFilter.cpp | 26 +- 6 files changed, 295 insertions(+), 217 deletions(-) diff --git a/bolt/connectors/paimon/PaimonDataSource.cpp b/bolt/connectors/paimon/PaimonDataSource.cpp index 23f569f2a..f431a932c 100644 --- a/bolt/connectors/paimon/PaimonDataSource.cpp +++ b/bolt/connectors/paimon/PaimonDataSource.cpp @@ -93,18 +93,32 @@ void collectFieldNames( const core::TypedExprPtr& expression, std::vector& fieldNames, std::unordered_set& seen) { - const auto* field = - dynamic_cast(expression.get()); - if (field && - (field->inputs().empty() || - dynamic_cast( - field->inputs().front().get()) != nullptr) && - seen.insert(field->name()).second) { - fieldNames.push_back(field->name()); + std::vector pending; + if (expression) { + pending.push_back(expression); } - for (const auto& input : expression->inputs()) { - collectFieldNames(input, fieldNames, seen); + 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]); + } } } diff --git a/bolt/connectors/paimon/PaimonFilterTranslator.cpp b/bolt/connectors/paimon/PaimonFilterTranslator.cpp index db5ef1c3b..5d1de0ad4 100644 --- a/bolt/connectors/paimon/PaimonFilterTranslator.cpp +++ b/bolt/connectors/paimon/PaimonFilterTranslator.cpp @@ -307,7 +307,7 @@ std::optional<::paimon::Literal> literalFromVector( auto ts = vector->as>()->valueAt(0); int64_t totalNanos = (ts.getSeconds() * 1'000'000'000LL) + ts.getNanos(); int64_t millis = totalNanos / 1'000'000; - int32_t nanoOfMillis = static_cast(totalNanos % 1'000'000); + auto nanoOfMillis = static_cast(totalNanos % 1'000'000); return ::paimon::Literal(::paimon::Timestamp(millis, nanoOfMillis)); } default: @@ -383,240 +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, evaluator); - 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, evaluator); - 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, evaluator); - } - 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, evaluator); - 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, evaluator); - 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"); - } - auto lit = extractLiteral(inputs[1], fieldType, evaluator); - if (!lit) { - return fail("lt: could not extract literal"); + // 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)); } - 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"); - } - auto lit = extractLiteral(inputs[1], fieldType, evaluator); - if (!lit) { - return fail("lte: 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::LessOrEqual( - fieldIndex, fieldName, fieldType, *lit)); - } - // 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"); + // 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)); } - 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"); + // 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)); } - auto lit = extractLiteral(inputs[1], fieldType, evaluator); - if (!lit) { - return fail("gte: could not extract literal"); + + // 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()); } - 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"); + // Is null: is_null(field) + if (opName == "is_null") { + return ok( + ::paimon::PredicateBuilder::IsNull(fieldIndex, fieldName, fieldType)); } - auto low = extractLiteral(inputs[1], fieldType, evaluator); - auto high = extractLiteral(inputs[2], fieldType, evaluator); - if (!low || !high) { - return fail("between: could not extract literals"); + + // Is not null: is_not_null(field) + if (opName == "is_not_null") { + return ok(::paimon::PredicateBuilder::IsNotNull( + fieldIndex, fieldName, fieldType)); } - 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"); + 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"); } - auto literals = std::move(result.value()); - if (negated) { - return ok(::paimon::PredicateBuilder::NotIn( - fieldIndex, fieldName, fieldType, literals)); + + 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; } - return ok(::paimon::PredicateBuilder::In( - fieldIndex, fieldName, fieldType, literals)); - } - // 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 (!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; } - if (fieldType != ::paimon::FieldType::STRING && - fieldType != ::paimon::FieldType::BINARY) { - return fail("like only supports string/binary fields"); + + if (results.size() < inputs.size()) { + return fail("compound predicate result stack underflow"); } - auto lit = extractLiteral(inputs[1], fieldType, evaluator); - if (!lit) { - return fail("like: could not extract pattern literal"); + 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])); } - auto result = ::paimon::PredicateBuilder::Like( - fieldIndex, fieldName, fieldType, *lit); + results.resize(firstChild); + + auto result = opName == "and" ? ::paimon::PredicateBuilder::And(children) + : ::paimon::PredicateBuilder::Or(children); if (!result.ok()) { - return fail("like: " + result.status().ToString()); + return fail( + opName == "and" ? "PredicateBuilder::And failed" + : "PredicateBuilder::Or failed"); } - return ok(std::move(result).value()); + results.push_back(std::move(result).value()); } - // Is null: is_null(field) - if (opName == "is_null") { - return ok( - ::paimon::PredicateBuilder::IsNull(fieldIndex, fieldName, fieldType)); + if (results.size() != 1) { + return fail("predicate translation produced an invalid result stack"); } - - // Is not null: is_not_null(field) - if (opName == "is_not_null") { - return ok(::paimon::PredicateBuilder::IsNotNull( - fieldIndex, fieldName, fieldType)); - } - - return fail(fmt::format("unsupported opName={}", opName)); + return ok(std::move(results.front())); } std::optional> diff --git a/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp index bb6ffdf3c..146e768a7 100644 --- a/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp +++ b/bolt/connectors/paimon/tests/PaimonFilterTranslatorTest.cpp @@ -496,7 +496,7 @@ TEST_F( auto it = filters.find(common::Subfield("payload")); ASSERT_NE(it, filters.end()); EXPECT_TRUE(it->second->testBytes("abc", 3)); - EXPECT_FALSE(it->second->testBytes("abd", 3)); + EXPECT_FALSE(it->second->testBytes("bcd", 3)); } // --------------------------------------------------------------------------- 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 70f21b153..9de40ba86 100644 --- a/bolt/expression/ExprToSubfieldFilter.cpp +++ b/bolt/expression/ExprToSubfieldFilter.cpp @@ -138,17 +138,27 @@ std::pair> toSubfieldFilter( void flattenTopLevelConjuncts( const core::TypedExprPtr& expr, std::vector& conjuncts) { - if (!expr) { - return; + std::vector pending; + if (expr) { + pending.push_back(expr); } - auto call = std::dynamic_pointer_cast(expr); - if (call && call->name() == "and") { - for (const auto& input : call->inputs()) { - flattenTopLevelConjuncts(input, conjuncts); + + 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; } - return; + conjuncts.push_back(std::move(current)); } - conjuncts.push_back(expr); } std::vector flattenTopLevelConjuncts(