hudi-agent commented on code in PR #19568: URL: https://github.com/apache/hudi/pull/19568#discussion_r3740646855
########## hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestIncrementalQueryProjection.java: ########## @@ -0,0 +1,163 @@ +/* + * 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.hudi.functional; + +import org.apache.hudi.DataSourceReadOptions; +import org.apache.hudi.DataSourceWriteOptions; +import org.apache.hudi.SparkAdapterSupport$; +import org.apache.hudi.common.config.HoodieMetadataConfig; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.testutils.SparkClientFunctionalTestHarness; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * An incremental query filters on {@code _hoodie_commit_time}, and that filter is pushed into the + * file reader. The reader evaluates a pushed predicate against the schema it was asked to read, so + * if the query's projection does not include the filtered column the predicate cannot be satisfied + * and every row is dropped -- silently, with no error. + * + * <p>That made {@code count()} and any projected read return zero rows on a CoW incremental query + * while {@code collect()} returned the right ones: {@code collect()} happens to project every + * column, including the filtered one, so it worked only incidentally. + * + * <p>Each case here asserts the same query returns the same rows regardless of what it projects. + */ +public class TestIncrementalQueryProjection extends SparkClientFunctionalTestHarness { + + private static StructType schema() { + return DataTypes.createStructType(new StructField[] { + DataTypes.createStructField("record_key", DataTypes.StringType, true), + DataTypes.createStructField("partition_path", DataTypes.StringType, true), + DataTypes.createStructField("payload", DataTypes.StringType, true) Review Comment: 🤖 The fix and all four tests are CoW-only. For MOR incremental, file slices that have log files route through the `HoodieFileGroupReader` path (the `case Some(fileSlice) if !isCount && (requiredSchema.nonEmpty || logFiles present)` branch) rather than `readBaseFile`, and there `requiredFilters` are handed to `SparkFileFormatInternalRowReaderContext` and applied against `getAppliedRequiredSchema(...)`. Have you checked whether a MOR incremental `count()`/projected read hits the same missing-`_hoodie_commit_time` predicate on the base-file read there? If so it may warrant either the same augmentation or at least a MOR test case; if it's already covered by the merge schema, a one-line note here would help. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala: ########## @@ -516,19 +516,73 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath: String, iter.map(mapper.apply(_)) } + /** + * Columns referenced by `filters` that are present in the table but missing from `schema`. + * + * A pushed-down filter is evaluated by the Parquet reader against the schema it was asked to + * read. If the filter references a column that schema does not contain, the predicate cannot be + * satisfied and every row is dropped -- silently, with no error. That is reachable whenever a + * caller supplies required filters independently of the projection: an incremental query always + * filters on _hoodie_commit_time, but a `count()` projects no columns at all and a + * `select(subset)` projects only what the user asked for. + * + * Candidates come from the full table schema rather than the requested one: on the failing path + * the requested schema is itself empty, so it cannot supply the column the filter needs. + * Partition columns are excluded by the caller, which appends them separately. + */ + private def filterColumnsMissingFrom(target: StructType, + candidates: StructType, Review Comment: 🤖 nit: `candidates` is a bit abstract here — the Javadoc calls it "the full table schema", so could you name the parameter `fullSchema` (or `tableSchema`) to make the call-site self-documenting? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
