[Feature] Support filter pushdown for Log Tables

[wuchong]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Currently, we have supported projection pushdown for Log Tables. Meanwhile, we want to support filter pushdown as well, which can improve query performance and resource efficiency a lot.

This feature is very similar to Kafka Header Filter, but header is not directly accessible in SQL. In SQL language, this is filter pushdown, that automatically optimizes users' where col1 = 'something' condition into a pushdown operation to avoid transferring the unnecessary data.

Solution

We may need to add some metadata/statistics (e.g., min/max) for each Arrow log batch. This allows TabletServers can apply the pushdowned predicate on the metadata without reading and deserializing the Arrow log.

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!

[kaori-seasons]

@wuchong Hi, does it mean that condition pushdown is supported in LogScan and the switch for condition pushdown is defined in flussapi.proto?

[sunxiaojian]

There should be changes in both writing and reading, and some statistical data should be recorded during writing to facilitate skip data during data reading

[michaelkoepf]

@wuchong @sunxiaojian

1. what is expected from a design and implementation perspective? storing the min and max value helps with range conditions on integer values. For point conditions, min-max only helps to some extent, i.e., deserialization can be skipped when the point condition is outside of the min-max range, otherwise we need to deserialize and check the data. for point conditions with other data types, especially with string values like the mentioned where col1 = 'something', we would need more sophisticated mechanisms like data sketches to know if we can skip it or need to deserialize it (sketches can give false positive results), right?
2. is my assumption correct that statistics would be stored alongside the schema as metadata in the arrow batch?

[sunxiaojian]

@wuchong @sunxiaojian

1. what is expected from a design and implementation perspective? storing the min and max value helps with range conditions on integer values. For point conditions, min-max only helps to some extent, i.e., deserialization can be skipped when the point condition is outside of the min-max range, otherwise we need to deserialize and check the data. for point conditions with other data types, especially with string values like the mentioned where col1 = 'something', we would need more sophisticated mechanisms like data sketches to know if we can skip it or need to deserialize it (sketches can give false positive results), right?
2. is my assumption correct that statistics would be stored alongside the schema as metadata in the arrow batch?

@michaelkoepf
I think this issue is expected to:

1. Accurately filter out data according to conditions, and implement pushdown to avoid returning a large amount of invalid data to the client.
2. Skip data in batches based on min/max or other accurate metadata information to reduce the scope of scanning. And statistics would be stored alongside the schema as metadata in the arrow batch.

For point queries, we do need a more complex indexing mechanism because min/max may have limited effectiveness in cases of data dispersion. However, I think we should first implement the basic functionality of min/max and pushdown, and then expand further later. @wuchong cc

[wuchong]

Hi @michaelkoepf ,

1. yes, min max would help in the case of filter pushdown, even for the string data types. However, we need a POC and comprehensive benchmarks to verify the idea.
2. yes, we need statistics to be stored alongside the arrow batch. The statistics format is an important part of the design because currently Arrow doesn't have statistics. The statistics need to be very efficient to be parsed, not too large, and better to be integrated into Arrow ecosystem. Maybe we need to invent an our own statistics format, maybe we can follow the statistics format proposed and in development in the Arrow community. This depends on the POC and benchmarks.

[sunxiaojian]

@wuchong If no one start this issue, you can assign it to me and I will start working on it.

[wuchong]

@sunxiaojian maybe we can discuss the design here first. The most important part of the design is the statistics format in LogRecordBatch and the filter pushdown performance and cost.

[sunxiaojian]

@sunxiaojian maybe we can discuss the design here first. The most important part of the design is the statistics format in LogRecordBatch and the filter pushdown performance and cost.

ok, I organize a design first and then send it out for discussion

[swuferhong]

I apologize for the late response. Below is my simple idea for implementing this issue, and hope it can be helpful.

1、Expand LogRecordBatch interface.

public interface LogRecordBatch {

    //Other methods that have already been implemented......
    
  
    
    /** The read context of a {@link LogRecordBatch} to read records. */
    interface ReadContext {
   
        //Other methods that have already been implemented......
        
        /**
         *  where clause, eg: a=100 and b> 10.
         */
        default Expression expression() {
            return null;
        }
    }


    /**
     * @return The statistics of this record batch
     */
    Statistics getStatistics(VectorSchemaRoot root, Expression expression);

    /**
     * The statistics of a {@link LogRecordBatch} which can be used for filtering records.
     */
    interface Statistics {

        /**
         * Get the minimum value of a specific column in the record batch.
         *
         * @param columnName The name of the column.
         * @return The minimum value of the column.
         */
        Object getMinValue(String columnName);

        /**
         * Get the maximum value of a specific column in the record batch.
         *
         * @param columnName The name of the column.
         * @return The maximum value of the column.
         */
        Object getMaxValue(String columnName);

        /**
         * Check if the record batch can be skipped based on the filter criteria.
         *
         * @param filter The filter criteria.
         * @return true if the record batch can be skipped, false otherwise.
         */
        boolean skip(Filter filter);
    }


    interface Filter {
        boolean filter(Expression expression, Statistics statistics);
    }

}

2、Add Expression interface.

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecord;
import com.alibaba.fluss.record.LogRecordBatch;
import com.alibaba.fluss.utils.Preconditions;

import java.io.Serializable;
import java.util.Locale;

public interface Expression extends Serializable {
  enum Operation {
    TRUE,
    FALSE,
    IS_NULL,
    NOT_NULL,
    IS_NAN,
    NOT_NAN,
    LT,
    LT_EQ,
    GT,
    GT_EQ,
    EQ,
    NOT_EQ,
    IN,
    NOT_IN,
    NOT,
    AND,
    OR,
    STARTS_WITH,
    NOT_STARTS_WITH,
    COUNT,
    COUNT_STAR,
    MAX,
    MIN;

    public static Operation fromString(String operationType) {
      Preconditions.checkArgument(null != operationType, "Invalid operation type: null");
      try {
        return Expression.Operation.valueOf(operationType.toUpperCase(Locale.ENGLISH));
      } catch (IllegalArgumentException e) {
        throw new IllegalArgumentException(
                String.format("Invalid operation type: %s", operationType), e);
      }
    }
  }

    /**
     * Returns the operation for an expression node.
     */
    Operation op();

    boolean accept(ExpressionVisitor visitor, LogRecord log);

    default boolean accept(ExpressionVisitor visitor, LogRecordBatch.Statistics statistics){
      return false;
    }

  }

3、 Add Expression implements e.g：GreaterThan/EqualTo/And/Not

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecord;

public class GreaterThan implements Expression {
    private final String column; // column
    private final Object value;  // column value

    public GreaterThan(String column, Object value) {
        super();
        this.column = column;
        this.value = value;
    }

    @Override
    public Operation op() {
        return Operation.GT;
    }

    public String getColumn() {
        return column;
    }

    public Object getValue() {
        return value;
    }

    @Override
    public boolean accept(ExpressionVisitor visitor, LogRecord log) {
        return visitor.greaterThan(this, log);
    }
}

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecord;

public class EqualTo implements Expression {
    private final String column; // column name
    private final Object value;  // column value.

    public EqualTo(String column, Object value) {
        this.column = column;
        this.value = value;
    }

    @Override
    public Operation op() {
        return Operation.EQ;
    }

    public String getColumn() {
        return column;
    }

    public Object getValue() {
        return value;
    }

    @Override
    public boolean accept(ExpressionVisitor visitor, LogRecord log) {
        return visitor.equalTo(this, log);
    }

}

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecord;

public class And implements Expression {
    private final Expression left;
    private final Expression right;

    public And(Expression left, Expression right) {
        this.left = left;
        this.right = right;
    }

    @Override
    public Operation op() {
        return Operation.AND;
    }

    public Expression getLeft() {
        return left;
    }

    public Expression getRight() {
        return right;
    }

    @Override
    public boolean accept(ExpressionVisitor visitor, LogRecord record) {
        return left.accept(visitor, record) && right.accept(visitor, record);
    }
}

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecord;

public class Not implements Expression {
    private final Expression expression;
    public Not(Expression expression) {
        this.expression = expression;
    }
    @Override
    public Operation op() {
        return Operation.NOT;
    }

    public Expression getExpression() {
        return expression;
    }

    @Override
    public boolean accept(ExpressionVisitor visitor, LogRecord record) {
        return !expression.accept(visitor, record);
    }
}

4、 Add ExpressionVisitor Interface.

package com.alibaba.fluss.record.filter;

public interface ExpressionVisitor<T> {
    boolean equalTo(EqualTo equalTo, T t);
    boolean greaterThan(GreaterThan greaterThan, T t);
    boolean and(And and, T t);
    boolean not(Not not, T t);

    // Others visitor.

}

5、 Add ExpressionVisitor implements. e.g : LogRecordExpressionVisitor/DataSkippingExpressionVisitor

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecordBatch;

// For min/max
public class DataSkippingExpressionVisitor implements ExpressionVisitor<LogRecordBatch.Statistics> {
    @Override
    public boolean equalTo(EqualTo equalTo, LogRecordBatch.Statistics statistics) {
        String column = equalTo.getColumn();
        Object value = equalTo.getValue();

        // compare value logic.

        throw new IllegalArgumentException("Unknown column: " + column);
    }

    @Override
    public boolean greaterThan(GreaterThan greaterThan, LogRecordBatch.Statistics statistics) {
        String column = greaterThan.getColumn();
        Object value = greaterThan.getValue();

        // compare value logic.

        throw new IllegalArgumentException("Unknown column: " + column);
    }

    @Override
    public boolean and(And and, LogRecordBatch.Statistics statistics) {
        return and.getLeft().accept(this, statistics) && and.getRight().accept(this, statistics);
    }

    @Override
    public boolean not(Not not, LogRecordBatch.Statistics statistics) {
        return !not.getExpression().accept(this, statistics);
    }
}

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecord;

// For log record
public class LogRecordExpressionVisitor implements ExpressionVisitor<LogRecord> {
    @Override
    public boolean equalTo(EqualTo equalTo,  LogRecord record) {
        String column = equalTo.getColumn();
        Object value = equalTo.getValue();

        // compare value logic.

        throw new IllegalArgumentException("Unknown column: " + column);
    }

    @Override
    public boolean greaterThan(GreaterThan greaterThan, LogRecord record) {
        String column = greaterThan.getColumn();
        Object value = greaterThan.getValue();

        // compare value logic.

        throw new IllegalArgumentException("Unknown column: " + column);
    }


    @Override
    public boolean and(And and, LogRecord record) {
        return and.getLeft().accept(this, record) && and.getRight().accept(this, record);
    }

    @Override
    public boolean not(Not not, LogRecord record) {
        return !not.getExpression().accept(this, record);
    }
}

6、 Add LogRecordBatch.Filter implements. e.g : DefaultLogFilter

package com.alibaba.fluss.record.filter;

import com.alibaba.fluss.record.LogRecordBatch;

public class DefaultLogFilter implements LogRecordBatch.Filter<LogRecordBatch.Statistics> {
    private ExpressionVisitor expressionVisitor;
    public DefaultLogFilter(ExpressionVisitor expressionVisitor){
        this.expressionVisitor = expressionVisitor;
    }
    @Override
    public boolean filter(Expression expression, LogRecordBatch.Statistics o) {
        return expression.accept(expressionVisitor, o);
    }
}

7、 Add LogRecordBatch.Statistics implements. e.g : DefaultStatistics

package com.alibaba.fluss.record;

import com.alibaba.fluss.record.filter.Expression;
import com.alibaba.fluss.shaded.arrow.org.apache.arrow.vector.VectorSchemaRoot;

public class DefaultStatistics implements LogRecordBatch.Statistics {
    private final VectorSchemaRoot root;
    private final Expression expression;
    public  DefaultStatistics(VectorSchemaRoot root, Expression expression){
        this.root = root;
        this.expression = expression;
    }
    /**
     * Get the minimum value of a specific column in the record batch.
     *
     * @param columnName The name of the column.
     * @return The minimum value of the column.
     */
    @Override
    public Object getMinValue(String columnName) {
        return null;
    }

    /**
     * Get the maximum value of a specific column in the record batch.
     *
     * @param columnName The name of the column.
     * @return The maximum value of the column.
     */
    @Override
    public Object getMaxValue(String columnName) {
        return null;
    }

    /**
     * Check if the record batch can be skipped based on the filter criteria.
     *
     * @param filter The filter criteria.
     * @return true if the record batch can be skipped, false otherwise.
     */
    @Override
    public boolean skip(LogRecordBatch.Filter filter) {
        return filter.filter(expression, this);
    }
}

8. DefaultLogRecordBatch Implementation Logic.

@PublicEvolving
public class DefaultLogRecordBatch implements LogRecordBatch {
   

    @Override
    public CloseableIterator<LogRecord> records(ReadContext context) {
        if (getRecordCount() == 0) {
            return CloseableIterator.emptyIterator();
        }

        int schemaId = schemaId();
        long timestamp = commitTimestamp();
        LogFormat logFormat = context.getLogFormat();
        RowType rowType = context.getRowType(schemaId);
        switch (logFormat) {
            case ARROW:
                return columnRecordIterator(
                        rowType,
                        context.getVectorSchemaRoot(schemaId),
                        context.getBufferAllocator(),
                        timestamp,
                        **context.expression());      // add new param.**
            case INDEXED:
                return rowRecordIterator(rowType, timestamp);
            default:
                throw new IllegalArgumentException("Unsupported log format: " + logFormat);
        }
    }

    /**
     * Get the statistics of this record batch. The statistics can be used for filtering records
     * based on certain criteria.
     *
     * @return The statistics of this record batch
     */
    @Override
    public Statistics getStatistics(VectorSchemaRoot root, Expression expression) {
        return new DefaultStatistics(root, expression);
    }


    private CloseableIterator<LogRecord> columnRecordIterator(
            RowType rowType, VectorSchemaRoot root, BufferAllocator allocator, long timestamp, Expression expression) {
        int rowKindOffset = position + ARROW_ROWKIND_OFFSET;
        RowKindVector rowKindVector = new RowKindVector(segment, rowKindOffset, getRecordCount());
        int arrowOffset = rowKindOffset + rowKindVector.sizeInBytes();
        int arrowLength = sizeInBytes() - ARROW_ROWKIND_OFFSET - rowKindVector.sizeInBytes();
        ArrowReader reader =
                ArrowUtils.createArrowReader(
                        segment, arrowOffset, arrowLength, root, allocator, rowType);

       Statistics statistics = getStatistics(root, expression);
       if (!statistics.skip(new DefaultLogFilter(new DataSkippingExpressionVisitor()))){

            // Skipping current batch data.

        }
       

        return new LogRecordIterator() {
            int rowId = 0;

            @Override
            public boolean hasNext() {
                return rowId < reader.getRowCount();
            }

            @Override
            protected LogRecord readNext(long baseOffset) {
                RowKind rowKind = rowKindVector.getRowKind(rowId);
                LogRecord record =
                        new GenericRecord(
                                baseOffset + rowId, timestamp, rowKind, reader.read(rowId));
                
                // Add filter log record.
                
                rowId++;
                return record;
            }

            @Override
            protected boolean ensureNoneRemaining() {
                return true;
            }

            @Override
            public void close() {
                reader.close();
            }
        };
    }

}

In summary, this is my simple idea of pseudocode, hoping to be helpful for implementation. Of course, there is also logic such as how Statistics is stored that has not been provided

Hi, @sunxiaojian . Thank you for the proposal you provided. This is a relatively complex task, and we will develop a preliminary plan based on our research into actual production needs. We will also validate the feasibility through benchmarking or some real production jobs. Once the preliminary plan is confirmed, we will share it under this issue for further discussion, possibly a Google Docs. If you would like to continue participating in this work, we can divide the tasks for implementation to move forward this job.