Copilot commented on code in PR #3501: URL: https://github.com/apache/fluss/pull/3501#discussion_r3445539390
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverter.java: ########## @@ -0,0 +1,281 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.lake.hudi.utils; + +import org.apache.fluss.predicate.And; +import org.apache.fluss.predicate.CompoundPredicate; +import org.apache.fluss.predicate.FieldRef; +import org.apache.fluss.predicate.FunctionVisitor; +import org.apache.fluss.predicate.LeafPredicate; +import org.apache.fluss.predicate.Or; +import org.apache.fluss.predicate.Predicate; +import org.apache.fluss.predicate.PredicateVisitor; +import org.apache.fluss.row.BinaryString; +import org.apache.fluss.row.TimestampLtz; +import org.apache.fluss.row.TimestampNtz; + +import org.apache.flink.table.expressions.FieldReferenceExpression; +import org.apache.flink.table.expressions.ValueLiteralExpression; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.hudi.org.apache.avro.Schema; +import org.apache.hudi.source.ExpressionPredicates; +import org.apache.hudi.util.AvroSchemaConverter; + +import java.time.ZoneOffset; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Converts Fluss predicates to Hudi expression predicates for Hudi reader pushdown. */ +public class FlussToHudiExpressionPredicateConverter + implements PredicateVisitor<ExpressionPredicates.Predicate> { + + private static final String HUDI_METADATA_COLUMN_PREFIX = "_hoodie_"; + + private final Schema hudiSchema; + private final int metadataFieldCount; + private final LeafFunctionConverter converter = new LeafFunctionConverter(); + + private FlussToHudiExpressionPredicateConverter(Schema hudiSchema) { + this.hudiSchema = hudiSchema; + this.metadataFieldCount = metadataFieldCount(hudiSchema); + } + + public static Optional<ExpressionPredicates.Predicate> convert( + Schema hudiSchema, Predicate flussPredicate) { + try { + return Optional.of( + flussPredicate.visit(new FlussToHudiExpressionPredicateConverter(hudiSchema))); + } catch (UnsupportedOperationException e) { + return Optional.empty(); + } + } + + @Override + public ExpressionPredicates.Predicate visit(LeafPredicate predicate) { + return predicate.visit(converter); + } + + @Override + public ExpressionPredicates.Predicate visit(CompoundPredicate predicate) { + List<ExpressionPredicates.Predicate> children = + predicate.children().stream().map(p -> p.visit(this)).collect(Collectors.toList()); + + CompoundPredicate.Function function = predicate.function(); + if (function instanceof And) { + return reducePredicates(children, true); + } else if (function instanceof Or) { + return reducePredicates(children, false); + } + + throw new UnsupportedOperationException( + "Unsupported Fluss compound predicate function: " + function); + } + + private ExpressionPredicates.Predicate reducePredicates( + List<ExpressionPredicates.Predicate> children, boolean and) { + if (children.isEmpty()) { + throw new UnsupportedOperationException("Empty compound predicate."); + } + + ExpressionPredicates.Predicate result = children.get(0); + for (int i = 1; i < children.size(); i++) { + result = + and + ? ExpressionPredicates.And.getInstance() + .bindPredicates(result, children.get(i)) + : ExpressionPredicates.Or.getInstance() + .bindPredicates(result, children.get(i)); + } + return result; + } + + private class LeafFunctionConverter implements FunctionVisitor<ExpressionPredicates.Predicate> { + + @Override + public ExpressionPredicates.Predicate visitIsNotNull(FieldRef fieldRef) { + throw new UnsupportedOperationException("Hudi does not support isNotNull pushdown."); + } + + @Override + public ExpressionPredicates.Predicate visitIsNull(FieldRef fieldRef) { + throw new UnsupportedOperationException("Hudi does not support isNull pushdown."); + } + + @Override + public ExpressionPredicates.Predicate visitStartsWith(FieldRef fieldRef, Object literal) { + throw new UnsupportedOperationException("Hudi does not support startsWith pushdown."); + } + + @Override + public ExpressionPredicates.Predicate visitEndsWith(FieldRef fieldRef, Object literal) { + throw new UnsupportedOperationException("Hudi does not support endsWith pushdown."); + } + + @Override + public ExpressionPredicates.Predicate visitContains(FieldRef fieldRef, Object literal) { + throw new UnsupportedOperationException("Hudi does not support contains pushdown."); + } + + @Override + public ExpressionPredicates.Predicate visitLessThan(FieldRef fieldRef, Object literal) { + return bind(ExpressionPredicates.LessThan.getInstance(), fieldRef, literal); + } + + @Override + public ExpressionPredicates.Predicate visitGreaterOrEqual( + FieldRef fieldRef, Object literal) { + return bind(ExpressionPredicates.GreaterThanOrEqual.getInstance(), fieldRef, literal); + } + + @Override + public ExpressionPredicates.Predicate visitNotEqual(FieldRef fieldRef, Object literal) { + return bind(ExpressionPredicates.NotEquals.getInstance(), fieldRef, literal); + } + + @Override + public ExpressionPredicates.Predicate visitLessOrEqual(FieldRef fieldRef, Object literal) { + return bind(ExpressionPredicates.LessThanOrEqual.getInstance(), fieldRef, literal); + } + + @Override + public ExpressionPredicates.Predicate visitEqual(FieldRef fieldRef, Object literal) { + return bind(ExpressionPredicates.Equals.getInstance(), fieldRef, literal); + } + + @Override + public ExpressionPredicates.Predicate visitGreaterThan(FieldRef fieldRef, Object literal) { + return bind(ExpressionPredicates.GreaterThan.getInstance(), fieldRef, literal); + } + + @Override + public ExpressionPredicates.Predicate visitIn(FieldRef fieldRef, List<Object> literals) { + Schema.Field field = getFieldByFlussIndex(fieldRef.index()); + DataType dataType = supportedHudiDataType(field); + List<ValueLiteralExpression> literalExpressions = + literals.stream() + .map(literal -> createValueLiteral(literal, dataType)) + .collect(Collectors.toList()); + return ExpressionPredicates.In.getInstance() + .bindValueLiterals(literalExpressions) + .bindFieldReference(createFieldReference(field, dataType)); + } + + @Override + public ExpressionPredicates.Predicate visitNotIn(FieldRef fieldRef, List<Object> literals) { + ExpressionPredicates.Predicate inPredicate = visitIn(fieldRef, literals); + return ExpressionPredicates.Not.getInstance().bindPredicate(inPredicate); + } + + @Override + public ExpressionPredicates.Predicate visitAnd( + List<ExpressionPredicates.Predicate> children) { + throw new UnsupportedOperationException("Unsupported visitAnd method."); + } + + @Override + public ExpressionPredicates.Predicate visitOr( + List<ExpressionPredicates.Predicate> children) { + throw new UnsupportedOperationException("Unsupported visitOr method."); + } + + private ExpressionPredicates.Predicate bind( + ExpressionPredicates.ColumnPredicate predicate, FieldRef fieldRef, Object literal) { + Schema.Field field = getFieldByFlussIndex(fieldRef.index()); + DataType dataType = supportedHudiDataType(field); + return predicate + .bindValueLiteral(createValueLiteral(literal, dataType)) + .bindFieldReference(createFieldReference(field, dataType)); + } + + private FieldReferenceExpression createFieldReference( + Schema.Field field, DataType dataType) { + return new FieldReferenceExpression(field.name(), dataType, 0, 0); + } Review Comment: `FieldReferenceExpression` is created with hard-coded indexes `(0, 0)`. In Flink expressions, the last parameter is the field index within the input; using `0` for all fields can bind predicates to the wrong column (especially with Hudi metadata columns prepended). Use the actual Avro field position instead. ########## fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java: ########## @@ -53,16 +48,14 @@ void testGetSplitSerializerReturnsHudiSplitSerializer() { } @Test - void testFiltersRemainWhenPushDownIsNotSupportedYet() { + void testWithEmptyFiltersDoesNotResolveHudiSchema() { HudiLakeSource source = new HudiLakeSource(new Configuration(), TablePath.of("db1", "table1")); - Predicate predicate = new PredicateBuilder(RowType.of(new IntType())).equal(0, 1); - LakeSource.FilterPushDownResult result = - source.withFilters(Collections.singletonList(predicate)); + LakeSource.FilterPushDownResult result = source.withFilters(Collections.emptyList()); Review Comment: The test name says it verifies that the Hudi schema is not resolved, but the test only asserts accepted/remaining predicates. Either add an assertion that schema resolution is skipped, or rename the test to match what it actually verifies to avoid misleading coverage. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSortedRecordReader.java: ########## @@ -0,0 +1,232 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.lake.hudi.source; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiTableInfo; +import org.apache.fluss.lake.source.SortedRecordReader; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.LogRecord; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypeRoot; +import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.CloseableIterator; +import org.apache.fluss.utils.InternalRowUtils; + +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.org.apache.avro.Schema; +import org.apache.hudi.source.ExpressionPredicates; +import org.apache.hudi.util.AvroSchemaConverter; +import org.apache.hudi.util.StreamerUtil; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; + +/** Hudi record reader that exposes primary-key ordering for Fluss union read. */ +public class HudiSortedRecordReader implements SortedRecordReader { + + private final Comparator<InternalRow> comparator; + private final @Nullable HudiRecordReader delegate; + + public HudiSortedRecordReader( + Configuration hudiConfig, + TablePath tablePath, + @Nullable HudiSplit hudiSplit, + @Nullable int[][] project, + List<ExpressionPredicates.Predicate> predicates) + throws Exception { + try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath, hudiConfig)) { + Schema avroSchema = + StreamerUtil.getTableAvroSchema(hudiTableInfo.getMetaClient(), true); + this.comparator = + createPrimaryKeyComparator( + avroSchema, primaryKeyFields(hudiTableInfo, tablePath)); + } + this.delegate = + hudiSplit == null + ? null + : new HudiRecordReader( + hudiConfig, tablePath, hudiSplit, project, predicates); + } + + @Override + public CloseableIterator<LogRecord> read() throws IOException { + return delegate == null ? CloseableIterator.emptyIterator() : delegate.read(); + } Review Comment: `SortedRecordReader` requires `read()` to return records in the order defined by `order()` (see SortedRecordReader Javadoc). This implementation only exposes a primary-key comparator but delegates reading to `HudiRecordReader` without enforcing/validating that the produced records are sorted by that comparator, which can break sort-merge union read correctness if the underlying Hudi iterator is not already ordered. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
